From 8aa1bbc64bdebcf11358121c18f51aeaf0b32b5a Mon Sep 17 00:00:00 2001 From: Carlos Santos <4a.santos@gmail.com> Date: Mon, 2 Jun 2025 15:41:37 +0200 Subject: [PATCH] Refactor storage service and interfaces for improved separation of concerns - Updated StorageFactory to create basic storage providers and key builders. - Simplified StorageProvider interface to focus on basic CRUD operations. - Enhanced MeetStorageService to handle domain-specific logic while delegating storage operations. - Implemented Redis caching for room data to improve performance. - Added error handling and logging improvements throughout the service. - Removed deprecated methods and streamlined object retrieval processes. refactor: update storage service and interfaces to include user key handling and improve initialization logic refactor: update beforeAll hooks in recording tests to clear rooms and recordings refactor: optimize integration recordings test command Revert "refactor: optimize integration recordings test command" This reverts commit d517a44fa282b91613f8c55130916c2af5f07267. refactor: enhance Redis cache storage operations refactor: streamline test setup and teardown for security and recordings APIs --- .../src/config/dependency-injector.config.ts | 26 +- backend/src/helpers/recording.helper.ts | 6 - backend/src/models/redis.model.ts | 2 + backend/src/services/index.ts | 2 +- .../src/services/livekit-webhook.service.ts | 2 - backend/src/services/recording.service.ts | 223 ++--- backend/src/services/redis.service.ts | 21 +- backend/src/services/room.service.ts | 26 +- backend/src/services/storage/index.ts | 3 +- .../storage/providers/s3-storage.provider.ts | 699 -------------- .../providers/s3/s3-storage-key.builder.ts | 41 + .../providers/s3/s3-storage.provider.ts | 140 +++ .../{ => storage/providers/s3}/s3.service.ts | 18 +- .../src/services/storage/storage.factory.ts | 43 +- .../src/services/storage/storage.interface.ts | 297 +++--- .../src/services/storage/storage.service.ts | 855 +++++++++++++----- .../api/global-preferences/security.test.ts | 15 +- .../recordings/bulk-delete-recording.test.ts | 3 +- .../api/recordings/delete-recording.test.ts | 7 +- .../recordings/get-media-recording.test.ts | 2 + .../api/recordings/get-recording.test.ts | 3 +- .../api/recordings/get-recordings.test.ts | 8 +- .../api/recordings/race-conditions.test.ts | 2 + .../api/recordings/start-recording.test.ts | 6 +- .../api/recordings/stop-recording.test.ts | 2 + .../integration/api/rooms/create-room.test.ts | 6 +- typings/src/auth-preferences.ts | 23 +- typings/src/global-preferences.ts | 16 +- typings/src/participant.ts | 32 +- typings/src/room-preferences.ts | 22 +- typings/src/user.ts | 12 +- 31 files changed, 1195 insertions(+), 1368 deletions(-) delete mode 100644 backend/src/services/storage/providers/s3-storage.provider.ts create mode 100644 backend/src/services/storage/providers/s3/s3-storage-key.builder.ts create mode 100644 backend/src/services/storage/providers/s3/s3-storage.provider.ts rename backend/src/services/{ => storage/providers/s3}/s3.service.ts (94%) diff --git a/backend/src/config/dependency-injector.config.ts b/backend/src/config/dependency-injector.config.ts index 45b5ea6..45662fd 100644 --- a/backend/src/config/dependency-injector.config.ts +++ b/backend/src/config/dependency-injector.config.ts @@ -14,14 +14,25 @@ import { S3Service, S3StorageProvider, StorageFactory, + StorageKeyBuilder, + StorageProvider, SystemEventService, TaskSchedulerService, TokenService, UserService } from '../services/index.js'; +import { MEET_PREFERENCES_STORAGE_MODE } from '../environment.js'; +import { S3KeyBuilder } from '../services/storage/providers/s3/s3-storage-key.builder.js'; export const container: Container = new Container(); +export const STORAGE_TYPES = { + StorageProvider: Symbol.for('StorageProvider'), + KeyBuilder: Symbol.for('KeyBuilder'), + S3StorageProvider: Symbol.for('S3StorageProvider'), + S3KeyBuilder: Symbol.for('S3KeyBuilder') +}; + /** * Registers all necessary dependencies in the container. * @@ -38,6 +49,7 @@ export const registerDependencies = () => { container.bind(MutexService).toSelf().inSingletonScope(); container.bind(TaskSchedulerService).toSelf().inSingletonScope(); + configureStorage(MEET_PREFERENCES_STORAGE_MODE); container.bind(S3Service).toSelf().inSingletonScope(); container.bind(S3StorageProvider).toSelf().inSingletonScope(); container.bind(StorageFactory).toSelf().inSingletonScope(); @@ -55,8 +67,20 @@ export const registerDependencies = () => { container.bind(LivekitWebhookService).toSelf().inSingletonScope(); }; +const configureStorage = (storageMode: string) => { + container.get(LoggerService).info(`Creating ${storageMode} storage provider`); + + switch (storageMode) { + default: + case 's3': + container.bind(STORAGE_TYPES.StorageProvider).to(S3StorageProvider).inSingletonScope(); + container.bind(STORAGE_TYPES.KeyBuilder).to(S3KeyBuilder).inSingletonScope(); + break; + } +}; + export const initializeEagerServices = async () => { // Force the creation of services that need to be initialized at startup container.get(RecordingService); - await container.get(MeetStorageService).initialize(); + await container.get(MeetStorageService).initializeGlobalPreferences(); }; diff --git a/backend/src/helpers/recording.helper.ts b/backend/src/helpers/recording.helper.ts index e7820ea..bb8cb5b 100644 --- a/backend/src/helpers/recording.helper.ts +++ b/backend/src/helpers/recording.helper.ts @@ -186,12 +186,6 @@ export class RecordingHelper { return size !== 0 ? size : undefined; } - static buildMetadataFilePath(recordingId: string): string { - const { roomId, egressId, uid } = RecordingHelper.extractInfoFromRecordingId(recordingId); - - return `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/.metadata/${roomId}/${egressId}/${uid}.json`; - } - private static toSeconds(nanoseconds: number): number { const nanosecondsToSeconds = 1 / 1_000_000_000; return nanoseconds * nanosecondsToSeconds; diff --git a/backend/src/models/redis.model.ts b/backend/src/models/redis.model.ts index a324f0c..8103023 100644 --- a/backend/src/models/redis.model.ts +++ b/backend/src/models/redis.model.ts @@ -5,6 +5,8 @@ export const enum RedisKeyPrefix { export const enum RedisKeyName { GLOBAL_PREFERENCES = `${RedisKeyPrefix.BASE}global_preferences`, ROOM = `${RedisKeyPrefix.BASE}room:`, + RECORDING = `${RedisKeyPrefix.BASE}recording:`, + ARCHIVED_ROOM = `${RedisKeyPrefix.BASE}archived_room:`, USER = `${RedisKeyPrefix.BASE}user:`, } diff --git a/backend/src/services/index.ts b/backend/src/services/index.ts index 7398e1a..ab650a6 100644 --- a/backend/src/services/index.ts +++ b/backend/src/services/index.ts @@ -4,7 +4,7 @@ export * from './system-event.service.js'; export * from './mutex.service.js'; export * from './task-scheduler.service.js'; -export * from './s3.service.js'; +export * from './storage/providers/s3/s3.service.js'; export * from './storage/index.js'; export * from './token.service.js'; diff --git a/backend/src/services/livekit-webhook.service.ts b/backend/src/services/livekit-webhook.service.ts index 5b665ea..c3d52e3 100644 --- a/backend/src/services/livekit-webhook.service.ts +++ b/backend/src/services/livekit-webhook.service.ts @@ -12,7 +12,6 @@ import { OpenViduWebhookService, RecordingService, RoomService, - S3Service, SystemEventService } from './index.js'; @@ -20,7 +19,6 @@ import { export class LivekitWebhookService { protected webhookReceiver: WebhookReceiver; constructor( - @inject(S3Service) protected s3Service: S3Service, @inject(RecordingService) protected recordingService: RecordingService, @inject(LiveKitService) protected livekitService: LiveKitService, @inject(RoomService) protected roomService: RoomService, diff --git a/backend/src/services/recording.service.ts b/backend/src/services/recording.service.ts index eb0ed7a..ff36a37 100644 --- a/backend/src/services/recording.service.ts +++ b/backend/src/services/recording.service.ts @@ -13,11 +13,9 @@ import { errorRecordingCannotBeStoppedWhileStarting, errorRecordingNotFound, errorRecordingNotStopped, - errorRecordingRangeNotSatisfiable, errorRecordingStartTimeout, errorRoomHasNoParticipants, errorRoomNotFound, - internalError, isErrorRecordingAlreadyStopped, isErrorRecordingCannotBeStoppedWhileStarting, isErrorRecordingNotFound, @@ -206,29 +204,16 @@ export class RecordingService { async deleteRecording(recordingId: string): Promise { try { // Get the recording metada and recording info from the S3 bucket - const { binaryFilesToDelete, metadataFilesToDelete, recordingInfo } = - await this.getDeletableRecordingFiles(recordingId); - const { roomId } = RecordingHelper.extractInfoFromRecordingId(recordingId); - const deleteRecordingTasks: Promise[] = []; + const { recordingInfo } = await this.storageService.getRecordingMetadata(recordingId); - if (binaryFilesToDelete.size > 0) { - // Delete video files from S3 - deleteRecordingTasks.push( - this.storageService.deleteRecordingBinaryFilesByPaths(Array.from(binaryFilesToDelete)) - ); - } + // Validate the recording status + if (!RecordingHelper.canBeDeleted(recordingInfo)) throw errorRecordingNotStopped(recordingId); - if (metadataFilesToDelete.size > 0) { - // Delete metadata files from storage provider - deleteRecordingTasks.push( - this.storageService.deleteRecordingMetadataByPaths(Array.from(metadataFilesToDelete)) - ); - } + await this.storageService.deleteRecording(recordingId); - await Promise.all(deleteRecordingTasks); - - this.logger.info(`Successfully deleted ${recordingId}`); + this.logger.info(`Successfully deleted recording ${recordingId}`); + const { roomId } = recordingInfo; const shouldDeleteRoomMetadata = await this.shouldDeleteRoomMetadata(roomId); if (shouldDeleteRoomMetadata) { @@ -253,8 +238,7 @@ export class RecordingService { async bulkDeleteRecordingsAndAssociatedFiles( recordingIds: string[] ): Promise<{ deleted: string[]; notDeleted: { recordingId: string; error: string }[] }> { - let allMetadataFilesToDelete: Set = new Set(); - let allBinaryFilesToDelete: Set = new Set(); + const validRecordingIds: Set = new Set(); const deletedRecordings: Set = new Set(); const notDeletedRecordings: Set<{ recordingId: string; error: string }> = new Set(); const roomsToCheck: Set = new Set(); @@ -262,35 +246,32 @@ export class RecordingService { // Check if the recording is in progress for (const recordingId of recordingIds) { try { - const { binaryFilesToDelete, metadataFilesToDelete } = - await this.getDeletableRecordingFiles(recordingId); - // Add files to the set of files to delete - allBinaryFilesToDelete = new Set([...allBinaryFilesToDelete, ...binaryFilesToDelete]); - allMetadataFilesToDelete = new Set([...allMetadataFilesToDelete, ...metadataFilesToDelete]); + const { recordingInfo } = await this.storageService.getRecordingMetadata(recordingId); + if (!RecordingHelper.canBeDeleted(recordingInfo)) { + throw errorRecordingNotStopped(recordingId); + } + + validRecordingIds.add(recordingId); deletedRecordings.add(recordingId); - // Track the roomId for checking if the room metadata file should be deleted - const { roomId } = RecordingHelper.extractInfoFromRecordingId(recordingId); - roomsToCheck.add(roomId); + // Track room for metadata cleanup + roomsToCheck.add(recordingInfo.roomId); } catch (error) { this.logger.error(`BulkDelete: Error processing recording ${recordingId}: ${error}`); notDeletedRecordings.add({ recordingId, error: (error as OpenViduMeetError).message }); } } - if (allBinaryFilesToDelete.size === 0) { + if (validRecordingIds.size === 0) { this.logger.warn(`BulkDelete: No eligible recordings found for deletion.`); return { deleted: Array.from(deletedRecordings), notDeleted: Array.from(notDeletedRecordings) }; } // Delete recordings and its metadata from S3 try { - await Promise.all([ - this.storageService.deleteRecordingBinaryFilesByPaths(Array.from(allBinaryFilesToDelete)), - this.storageService.deleteRecordingMetadataByPaths(Array.from(allMetadataFilesToDelete)) - ]); - this.logger.info(`BulkDelete: Successfully deleted ${allBinaryFilesToDelete.size} recordings.`); + await this.storageService.deleteRecordings(Array.from(validRecordingIds)); + this.logger.info(`BulkDelete: Successfully deleted ${validRecordingIds.size} recordings.`); } catch (error) { this.logger.error(`BulkDelete: Error performing bulk deletion: ${error}`); throw error; @@ -298,27 +279,35 @@ export class RecordingService { // Check if the room metadata file should be deleted const roomMetadataToDelete: string[] = []; - const deleteTasks: Promise[] = []; for (const roomId of roomsToCheck) { const shouldDeleteRoomMetadata = await this.shouldDeleteRoomMetadata(roomId); if (shouldDeleteRoomMetadata) { - deleteTasks.push(this.storageService.deleteArchivedRoomMetadata(roomId)); roomMetadataToDelete.push(roomId); } } + if (roomMetadataToDelete.length === 0) { + this.logger.verbose(`BulkDelete: No room metadata files to delete.`); + return { deleted: Array.from(deletedRecordings), notDeleted: Array.from(notDeletedRecordings) }; + } + + // Perform bulk deletion of room metadata files try { - this.logger.verbose(`Deleting room_metadata.json for rooms: ${roomMetadataToDelete.join(', ')}`); - await Promise.all(deleteTasks); + await Promise.all( + roomMetadataToDelete.map((roomId) => this.storageService.deleteArchivedRoomMetadata(roomId)) + ); this.logger.verbose(`BulkDelete: Successfully deleted ${roomMetadataToDelete.length} room metadata files.`); } catch (error) { this.logger.error(`BulkDelete: Error performing bulk deletion: ${error}`); throw error; } - return { deleted: Array.from(deletedRecordings), notDeleted: Array.from(notDeletedRecordings) }; + return { + deleted: Array.from(deletedRecordings), + notDeleted: Array.from(notDeletedRecordings) + }; } /** @@ -330,11 +319,10 @@ export class RecordingService { */ protected async shouldDeleteRoomMetadata(roomId: string): Promise { try { - const metadataPrefix = `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/.metadata/${roomId}`; - const { Contents } = await this.storageService.listObjects(metadataPrefix, 1); + const { recordings } = await this.storageService.getAllRecordings(roomId, 1); - // If no metadata files exist or the list is empty, the room metadata should be deleted - return !Contents || Contents.length === 0; + // If no recordings exist or the list is empty, the room metadata should be deleted + return !recordings || recordings.length === 0; } catch (error) { this.logger.warn(`Error checking room metadata for deletion (room ${roomId}): ${error}`); return null; @@ -363,45 +351,32 @@ export class RecordingService { * - `nextPageToken`: (Optional) A token to retrieve the next page of results, if available. * @throws Will throw an error if there is an issue retrieving the recordings. */ - async getAllRecordings({ maxItems, nextPageToken, roomId, fields }: MeetRecordingFilters): Promise<{ + async getAllRecordings(filters: MeetRecordingFilters): Promise<{ recordings: MeetRecordingInfo[]; isTruncated: boolean; nextPageToken?: string; }> { try { - // Construct the room prefix if a room ID is provided - const roomPrefix = roomId ? `/${roomId}` : ''; - const recordingPrefix = `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/.metadata${roomPrefix}`; + const { maxItems, nextPageToken, roomId, fields } = filters; - // Retrieve the recordings from the S3 bucket - const { Contents, IsTruncated, NextContinuationToken } = await this.storageService.listObjects( - recordingPrefix, - maxItems, - nextPageToken - ); + const response = await this.storageService.getAllRecordings(roomId, maxItems, nextPageToken); - if (!Contents) { - this.logger.verbose('No recordings found. Returning an empty array.'); - return { recordings: [], isTruncated: false }; + // Apply field filtering if specified + if (fields) { + response.recordings = response.recordings.map((rec) => + UtilsHelper.filterObjectFields(rec, fields) + ) as MeetRecordingInfo[]; } - const promises: Promise[] = []; - // Retrieve the metadata for each recording - Contents.forEach((item) => { - if (item?.Key && item.Key.endsWith('.json') && !item.Key.endsWith('secrets.json')) { - promises.push( - this.storageService.getRecordingMetadataByPath(item.Key) as Promise - ); - } - }); - - let recordings = await Promise.all(promises); - - recordings = recordings.map((rec) => UtilsHelper.filterObjectFields(rec, fields)) as MeetRecordingInfo[]; + const { recordings, isTruncated, nextContinuationToken } = response; this.logger.info(`Retrieved ${recordings.length} recordings.`); // Return the paginated list of recordings - return { recordings, isTruncated: !!IsTruncated, nextPageToken: NextContinuationToken }; + return { + recordings, + isTruncated: Boolean(isTruncated), + nextPageToken: nextContinuationToken + }; } catch (error) { this.logger.error(`Error getting recordings: ${error}`); throw error; @@ -410,64 +385,33 @@ export class RecordingService { async getRecordingAsStream( recordingId: string, - range?: string + rangeHeader?: string ): Promise<{ fileSize: number | undefined; fileStream: Readable; start?: number; end?: number }> { - const DEFAULT_RECORDING_FILE_PORTION_SIZE = 5 * 1024 * 1024; // 5MB + const DEFAULT_CHUNK_SIZE = 5 * 1024 * 1024; // 5MB + + // Ensure the recording is streamable const recordingInfo: MeetRecordingInfo = await this.getRecording(recordingId); if (recordingInfo.status !== MeetRecordingStatus.COMPLETE) { throw errorRecordingNotStopped(recordingId); } - const recordingPath = `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/${RecordingHelper.extractFilename(recordingInfo)}`; + let validatedRange = undefined; - if (!recordingPath) throw new Error(`Error extracting path from recording ${recordingId}`); + // Parse the range header if provided + if (rangeHeader) { + const match = rangeHeader.match(/^bytes=(\d+)-(\d*)$/)!; + const endStr = match[2]; - const { contentLength: fileSize } = await this.storageService.getObjectHeaders(recordingPath); - - if (!fileSize) { - this.logger.error(`Error getting file size for recording ${recordingId}`); - throw internalError(`getting file size for recording '${recordingId}'`); - } - - if (range) { - // Parse the range header - const matches = range.match(/^bytes=(\d+)-(\d*)$/)!; - - const start = parseInt(matches[1], 10); - let end = matches[2] ? parseInt(matches[2], 10) : start + DEFAULT_RECORDING_FILE_PORTION_SIZE; - - // Validate the range values - if (isNaN(start) || isNaN(end) || start < 0) { - this.logger.warn(`Invalid range values for recording ${recordingId}: start=${start}, end=${end}`); - this.logger.warn(`Returning full stream for recording ${recordingId}`); - return this.getFullStreamResponse(recordingPath, fileSize); - } - - if (start >= fileSize) { - this.logger.error( - `Invalid range values for recording ${recordingId}: start=${start}, end=${end}, fileSize=${fileSize}` - ); - throw errorRecordingRangeNotSatisfiable(recordingId, fileSize); - } - - // Adjust the end value to ensure it doesn't exceed the file size - end = Math.min(end, fileSize - 1); - - // If the start is greater than the end, return the full stream - if (start > end) { - this.logger.warn(`Invalid range values after adjustment: start=${start}, end=${end}`); - return this.getFullStreamResponse(recordingPath, fileSize); - } - - const fileStream = await this.storageService.getRecordingMedia(recordingPath, { - start, - end - }); - return { fileSize, fileStream, start, end }; + const start = parseInt(match[1], 10); + const end = endStr ? parseInt(endStr, 10) : start + DEFAULT_CHUNK_SIZE - 1; + validatedRange = { start, end }; + this.logger.debug(`Streaming partial content for recording '${recordingId}' from ${start} to ${end}.`); } else { - return this.getFullStreamResponse(recordingPath, fileSize); + this.logger.debug(`Streaming full content for recording '${recordingId}'.`); } + + return this.storageService.getRecordingMedia(recordingId, validatedRange); } protected async validateRoomForStartRecording(roomId: string): Promise { @@ -486,14 +430,6 @@ export class RecordingService { if (!hasParticipants) throw errorRoomHasNoParticipants(roomId); } - protected async getFullStreamResponse( - recordingPath: string, - fileSize: number - ): Promise<{ fileSize: number; fileStream: Readable }> { - const fileStream = await this.storageService.getRecordingMedia(recordingPath); - return { fileSize, fileStream }; - } - /** * Acquires a Redis-based lock to indicate that a recording is active for a specific room. * @@ -563,37 +499,6 @@ export class RecordingService { } } - /** - * Retrieves the data required to delete a recording, including the file paths - * to be deleted and the recording's metadata information. - * - * @param recordingId - The unique identifier of the recording egress. - */ - protected async getDeletableRecordingFiles(recordingId: string): Promise<{ - binaryFilesToDelete: Set; - metadataFilesToDelete: Set; - recordingInfo: MeetRecordingInfo; - }> { - const { metadataFilePath, recordingInfo } = await this.storageService.getRecordingMetadata(recordingId); - const binaryFilesToDelete: Set = new Set(); - const metadataFilesToDelete: Set = new Set(); - - // Validate the recording status - if (!RecordingHelper.canBeDeleted(recordingInfo)) throw errorRecordingNotStopped(recordingId); - - const filename = RecordingHelper.extractFilename(recordingInfo); - - if (!filename) { - throw internalError(`extracting path from recording '${recordingId}'`); - } - - const recordingPath = `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/${filename}`; - binaryFilesToDelete.add(recordingPath); - metadataFilesToDelete.add(metadataFilePath); - - return { binaryFilesToDelete, metadataFilesToDelete, recordingInfo }; - } - protected generateCompositeOptionsFromRequest(layout = 'grid'): RoomCompositeOptions { return { layout: layout diff --git a/backend/src/services/redis.service.ts b/backend/src/services/redis.service.ts index 4113d74..7d110fa 100644 --- a/backend/src/services/redis.service.ts +++ b/backend/src/services/redis.service.ts @@ -222,9 +222,11 @@ export class RedisService extends EventEmitter { } /** - * Deletes a key from Redis. - * @param key - The key to delete. - * @returns A promise that resolves to the number of keys deleted. + * Deletes one or more keys from Redis. + * + * @param keys - A single key string or an array of key strings to delete from Redis + * @returns A Promise that resolves to the number of keys that were successfully deleted + * @throws {Error} Throws an internal error if the deletion operation fails */ delete(keys: string | string[]): Promise { try { @@ -238,8 +240,19 @@ export class RedisService extends EventEmitter { } } - quit() { + cleanup() { + this.logger.verbose('Cleaning up Redis connections'); this.redisPublisher.quit(); + this.redisSubscriber.quit(); + this.removeAllListeners(); + + if (this.eventHandler) { + this.off('systemEvent', this.eventHandler); + this.eventHandler = undefined; + } + + this.isConnected = false; + this.logger.verbose('Redis connections cleaned up'); } async checkHealth() { diff --git a/backend/src/services/room.service.ts b/backend/src/services/room.service.ts index f4dd59a..03c1be3 100644 --- a/backend/src/services/room.service.ts +++ b/backend/src/services/room.service.ts @@ -13,7 +13,12 @@ import { uid as secureUid } from 'uid/secure'; import { uid } from 'uid/single'; import INTERNAL_CONFIG from '../config/internal-config.js'; import { MeetRoomHelper, OpenViduComponentsAdapterHelper, UtilsHelper } from '../helpers/index.js'; -import { errorInvalidRoomSecret, errorRoomMetadataNotFound, internalError } from '../models/error.model.js'; +import { + errorInvalidRoomSecret, + errorRoomMetadataNotFound, + errorRoomNotFound, + internalError +} from '../models/error.model.js'; import { IScheduledTask, LiveKitService, @@ -126,7 +131,7 @@ export class RoomService { await this.storageService.saveMeetRoom(room); // Update the archived room metadata if it exists - await this.storageService.updateArchivedRoomMetadata(roomId); + await this.storageService.archiveRoomMetadata(roomId); return room; } @@ -160,8 +165,10 @@ export class RoomService { }> { const response = await this.storageService.getMeetRooms(maxItems, nextPageToken); - const filteredRooms = response.rooms.map((room) => UtilsHelper.filterObjectFields(room, fields)); - response.rooms = filteredRooms as MeetRoom[]; + if (fields) { + const filteredRooms = response.rooms.map((room: MeetRoom) => UtilsHelper.filterObjectFields(room, fields)); + response.rooms = filteredRooms as MeetRoom[]; + } return response; } @@ -175,6 +182,11 @@ export class RoomService { async getMeetRoom(roomId: string, fields?: string): Promise { const meetRoom = await this.storageService.getMeetRoom(roomId); + if (!meetRoom) { + this.logger.error(`Meet room with ID ${roomId} not found.`); + throw errorRoomNotFound(roomId); + } + return UtilsHelper.filterObjectFields(meetRoom, fields) as MeetRoom; } @@ -251,6 +263,12 @@ export class RoomService { */ protected async markRoomAsDeleted(roomId: string): Promise { const room = await this.storageService.getMeetRoom(roomId); + + if (!room) { + this.logger.error(`Room with ID ${roomId} not found for deletion.`); + throw errorRoomNotFound(roomId); + } + room.markedForDeletion = true; await this.storageService.saveMeetRoom(room); } diff --git a/backend/src/services/storage/index.ts b/backend/src/services/storage/index.ts index b711372..e9043ca 100644 --- a/backend/src/services/storage/index.ts +++ b/backend/src/services/storage/index.ts @@ -1,4 +1,5 @@ export * from './storage.interface.js'; -export * from './providers/s3-storage.provider.js'; export * from './storage.factory.js'; export * from './storage.service.js'; + +export * from './providers/s3/s3-storage.provider.js'; diff --git a/backend/src/services/storage/providers/s3-storage.provider.ts b/backend/src/services/storage/providers/s3-storage.provider.ts deleted file mode 100644 index 395f31e..0000000 --- a/backend/src/services/storage/providers/s3-storage.provider.ts +++ /dev/null @@ -1,699 +0,0 @@ -import { PutObjectCommandOutput } from '@aws-sdk/client-s3'; -import { GlobalPreferences, MeetRecordingInfo, MeetRoom, User } from '@typings-ce'; -import { inject, injectable } from 'inversify'; -import INTERNAL_CONFIG from '../../../config/internal-config.js'; -import { errorRecordingNotFound, OpenViduMeetError, RedisKeyName } from '../../../models/index.js'; -import { LoggerService, RedisService, S3Service, StorageProvider } from '../../index.js'; -import { RecordingHelper } from '../../../helpers/recording.helper.js'; -import { Readable } from 'stream'; - -/** - * Implementation of the StorageProvider interface using AWS S3 for persistent storage - * with Redis caching for improved performance. - * - * This class provides operations for storing and retrieving application preferences, - * rooms, recordings metadata and users with a two-tiered storage approach: - * - Redis is used as a primary cache for fast access - * - S3 serves as the persistent storage layer and fallback when data is not in Redis - * - * The storage operations are performed in parallel to both systems when writing data, - * with transaction-like rollback behavior if one operation fails. - * - * @template GPrefs - Type for global preferences data, defaults to GlobalPreferences - * @template MRoom - Type for room data, defaults to MeetRoom - * @template MRec - Type for recording metadata, defaults to MeetRecordingInfo - * @template MUser - Type for user data, defaults to User - * - * @implements {StorageProvider} - */ -@injectable() -export class S3StorageProvider< - GPrefs extends GlobalPreferences = GlobalPreferences, - MRoom extends MeetRoom = MeetRoom, - MRec extends MeetRecordingInfo = MeetRecordingInfo, - MUser extends User = User -> implements StorageProvider -{ - protected readonly S3_GLOBAL_PREFERENCES_KEY = `global-preferences.json`; - - constructor( - @inject(LoggerService) protected logger: LoggerService, - @inject(S3Service) protected s3Service: S3Service, - @inject(RedisService) protected redisService: RedisService - ) {} - - /** - * Retrieves metadata headers for an object stored in S3. - * - * @param filePath - The path/key of the file in the S3 bucket - * @returns A promise that resolves to an object containing the content length and content type of the file - * @throws Will throw an error if the S3 operation fails or the file doesn't exist - */ - async getObjectHeaders(filePath: string): Promise<{ contentLength?: number; contentType?: string }> { - try { - const data = await this.s3Service.getHeaderObject(filePath); - return { - contentLength: data.ContentLength, - contentType: data.ContentType - }; - } catch (error) { - this.logger.error(`Error fetching object headers for ${filePath}: ${error}`); - throw error; - } - } - - /** - * Lists objects in the storage with optional pagination support. - * - * @param prefix - The prefix to filter objects by (acts as a folder path) - * @param maxItems - Maximum number of items to return (optional) - * @param nextPageToken - Token for pagination to get the next page (optional) - * @returns Promise resolving to paginated list of objects with metadata - */ - async listObjects( - prefix: string, - maxItems?: number, - nextPageToken?: string - ): Promise<{ - Contents?: Array<{ - Key?: string; - LastModified?: Date; - Size?: number; - ETag?: string; - }>; - IsTruncated?: boolean; - NextContinuationToken?: string; - }> { - try { - this.logger.debug( - `Listing objects with prefix: ${prefix}, maxItems: ${maxItems}, nextPageToken: ${nextPageToken}` - ); - return await this.s3Service.listObjectsPaginated(prefix, maxItems, nextPageToken); - } catch (error) { - this.handleError(error, `Error listing objects with prefix ${prefix}`); - throw error; - } - } - - /** - * Initializes global preferences. If no preferences exist, persists the provided defaults. - * If preferences exist but belong to a different project, they are replaced. - * - * @param defaultPreferences - The default preferences to initialize with. - */ - async initialize(defaultPreferences: GPrefs): Promise { - try { - const existingPreferences = await this.getGlobalPreferences(); - - if (!existingPreferences) { - this.logger.info('No existing preferences found. Saving default preferences to S3.'); - await this.saveGlobalPreferences(defaultPreferences); - return; - } - - this.logger.verbose('Global preferences found. Checking project association...'); - const isDifferentProject = existingPreferences.projectId !== defaultPreferences.projectId; - - if (isDifferentProject) { - this.logger.warn( - `Existing global preferences belong to project [${existingPreferences.projectId}], ` + - `which differs from current project [${defaultPreferences.projectId}]. Replacing preferences.` - ); - - await this.saveGlobalPreferences(defaultPreferences); - return; - } - - this.logger.verbose( - 'Global preferences for the current project are already initialized. No action needed.' - ); - } catch (error) { - this.logger.error('Error during global preferences initialization:', error); - } - } - - /** - * Retrieves the global preferences. - * First attempts to retrieve from Redis; if not available, falls back to S3. - * If fetched from S3, caches the result in Redis. - * - * @returns A promise that resolves to the global preferences or null if not found. - */ - async getGlobalPreferences(): Promise { - try { - // Try to get preferences from Redis cache - let preferences: GPrefs | null = await this.getFromRedis(RedisKeyName.GLOBAL_PREFERENCES); - - if (!preferences) { - this.logger.debug('Global preferences not found in Redis. Fetching from S3...'); - preferences = await this.getFromS3(this.S3_GLOBAL_PREFERENCES_KEY); - - if (preferences) { - this.logger.verbose('Fetched global preferences from S3. Caching them in Redis.'); - const redisPayload = JSON.stringify(preferences); - await this.redisService.set(RedisKeyName.GLOBAL_PREFERENCES, redisPayload, false); - } else { - this.logger.warn('No global preferences found in S3.'); - } - } else { - this.logger.verbose('Global preferences retrieved from Redis.'); - } - - return preferences; - } catch (error) { - this.handleError(error, 'Error fetching preferences'); - return null; - } - } - - /** - * Persists the global preferences to both S3 and Redis in parallel. - * Uses Promise.all to execute both operations concurrently. - * - * @param preferences - Global preferences to store. - * @returns The saved preferences. - * @throws Rethrows any error if saving fails. - */ - async saveGlobalPreferences(preferences: GPrefs): Promise { - try { - const redisPayload = JSON.stringify(preferences); - - await Promise.all([ - this.s3Service.saveObject(this.S3_GLOBAL_PREFERENCES_KEY, preferences), - this.redisService.set(RedisKeyName.GLOBAL_PREFERENCES, redisPayload, false) - ]); - this.logger.info('Global preferences saved successfully'); - return preferences; - } catch (error) { - this.handleError(error, 'Error saving global preferences'); - throw error; - } - } - - /** - * Persists a room object to S3 and Redis concurrently. - * If at least one operation fails, performs a rollback by deleting the successfully saved object. - * - * @param meetRoom - The room object to save. - * @returns The saved room if both operations succeed. - * @throws The error from the first failed operation. - */ - async saveMeetRoom(meetRoom: MRoom): Promise { - const { roomId } = meetRoom; - const s3Path = `${INTERNAL_CONFIG.S3_ROOMS_PREFIX}/${roomId}/${roomId}.json`; - const redisPayload = JSON.stringify(meetRoom); - const redisKey = RedisKeyName.ROOM + roomId; - - const [s3Result, redisResult] = await Promise.allSettled([ - this.s3Service.saveObject(s3Path, meetRoom), - this.redisService.set(redisKey, redisPayload, false) - ]); - - if (s3Result.status === 'fulfilled' && redisResult.status === 'fulfilled') { - return meetRoom; - } - - // Rollback any changes made by the successful operation - await this.rollbackRoomSave(roomId, s3Result, redisResult, s3Path, redisKey); - - // Return the error that occurred first - const failedOperation: PromiseRejectedResult = - s3Result.status === 'rejected' ? s3Result : (redisResult as PromiseRejectedResult); - const error = failedOperation.reason; - this.handleError(error, `Error saving Room preferences for room ${roomId}`); - throw error; - } - - /** - * Retrieves the list of Meet rooms from S3. - * - * @param maxItems - Maximum number of items to retrieve. - * @param nextPageToken - Continuation token for pagination. - * @returns An object containing the list of rooms, a flag indicating whether the list is truncated, and, if available, the next page token. - */ - async getMeetRooms( - maxItems: number, - nextPageToken?: string - ): Promise<{ - rooms: MRoom[]; - isTruncated: boolean; - nextPageToken?: string; - }> { - try { - const { - Contents: roomFiles, - IsTruncated, - NextContinuationToken - } = await this.s3Service.listObjectsPaginated(INTERNAL_CONFIG.S3_ROOMS_PREFIX, maxItems, nextPageToken); - - if (!roomFiles || roomFiles.length === 0) { - this.logger.verbose('No room files found in S3.'); - return { rooms: [], isTruncated: false }; - } - - // Extract room IDs directly and filter out invalid values - const roomIds = roomFiles - .map((file) => this.extractRoomId(file.Key)) - .filter((id): id is string => Boolean(id)); - - // Fetch and log any room lookup errors individually - // Fetch room preferences in parallel - const rooms = await Promise.all( - roomIds.map(async (roomId) => { - try { - return await this.getMeetRoom(roomId); - } catch (error: unknown) { - this.logger.warn(`Failed to fetch room "${roomId}": ${error}`); - return null; - } - }) - ); - - // Filter out null values - const validRooms = rooms.filter((room) => room !== null) as MRoom[]; - return { rooms: validRooms, isTruncated: !!IsTruncated, nextPageToken: NextContinuationToken }; - } catch (error) { - this.handleError(error, 'Error fetching Room preferences'); - return { rooms: [], isTruncated: false }; - } - } - - async getMeetRoom(roomId: string): Promise { - try { - // Try to get room preferences from Redis cache - const room: MRoom | null = await this.getFromRedis(roomId); - - if (!room) { - const s3RoomPath = `${INTERNAL_CONFIG.S3_ROOMS_PREFIX}/${roomId}/${roomId}.json`; - this.logger.debug(`Room ${roomId} not found in Redis. Fetching from S3 at ${s3RoomPath}...`); - - return await this.getFromS3(s3RoomPath); - } - - this.logger.debug(`Room ${roomId} verified in Redis`); - return room; - } catch (error) { - this.handleError(error, `Error fetching Room preferences for room ${roomId}`); - return null; - } - } - - async deleteMeetRooms(roomIds: string[]): Promise { - const roomsToDelete = roomIds.map((id) => `${INTERNAL_CONFIG.S3_ROOMS_PREFIX}/${id}/${id}.json`); - const redisKeysToDelete = roomIds.map((id) => RedisKeyName.ROOM + id); - - try { - await Promise.all([ - this.s3Service.deleteObjects(roomsToDelete), - this.redisService.delete(redisKeysToDelete) - ]); - this.logger.verbose(`Rooms deleted successfully: ${roomIds.join(', ')}`); - } catch (error) { - this.handleError(error, `Error deleting rooms: ${roomIds.join(', ')}`); - } - } - - async getArchivedRoomMetadata(roomId: string): Promise | null> { - try { - const filePath = `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/.room_metadata/${roomId}/room_metadata.json`; - const roomMetadata = await this.getFromS3>(filePath); - - if (!roomMetadata) { - this.logger.warn(`Room metadata not found for room ${roomId} in recordings bucket`); - return null; - } - - return roomMetadata; - } catch (error) { - this.handleError(error, `Error fetching archived room metadata for room ${roomId}`); - return null; - } - } - - /** - * Saves room metadata to a JSON file in the S3 bucket if it doesn't already exist. - * - * This method checks if the metadata file for the given room already exists in the - * S3 bucket. If not, it retrieves the room information, extracts the necessary - * secrets and preferences, and saves them to a metadata JSON file in the - * .room_metadata/{roomId}/ directory of the S3 bucket. - * - * @param roomId - The unique identifier of the room - */ - async archiveRoomMetadata(roomId: string): Promise { - try { - const filePath = `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/.room_metadata/${roomId}/room_metadata.json`; - const fileExists = await this.s3Service.exists(filePath); - - if (fileExists) { - this.logger.debug(`Room metadata already saved for room ${roomId} in recordings bucket`); - return; - } - - const room = await this.getMeetRoom(roomId); - - if (room) { - const roomMetadata = { - moderatorRoomUrl: room.moderatorRoomUrl, - publisherRoomUrl: room.publisherRoomUrl, - preferences: { - recordingPreferences: room.preferences?.recordingPreferences - } - }; - await this.s3Service.saveObject(filePath, roomMetadata); - this.logger.debug(`Room metadata saved for room ${roomId} in recordings bucket`); - return; - } - - this.logger.error(`Error saving room metadata for room ${roomId} in recordings bucket`); - } catch (error) { - this.logger.error(`Error saving room metadata for room ${roomId} in recordings bucket: ${error}`); - } - } - - /** - * Updates the archived room metadata for a given room in the S3 recordings bucket if it exists. - * - * @param roomId - The unique identifier of the room whose metadata needs to be updated. - */ - async updateArchivedRoomMetadata(roomId: string): Promise { - try { - const filePath = `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/.room_metadata/${roomId}/room_metadata.json`; - const fileExists = await this.s3Service.exists(filePath); - - if (!fileExists) { - this.logger.warn(`Room metadata not found for room ${roomId} in recordings bucket`); - return; - } - - const room = await this.getMeetRoom(roomId); - - if (room) { - const roomMetadata = { - moderatorRoomUrl: room.moderatorRoomUrl, - publisherRoomUrl: room.publisherRoomUrl, - preferences: { - recordingPreferences: room.preferences?.recordingPreferences - } - }; - await this.s3Service.saveObject(filePath, roomMetadata); - this.logger.debug(`Room metadata updated for room ${roomId} in recordings bucket`); - return; - } - - this.logger.error(`Error updating room metadata for room ${roomId} in recordings bucket`); - } catch (error) { - this.logger.error(`Error updating room metadata for room ${roomId} in recordings bucket: ${error}`); - } - } - - async deleteArchivedRoomMetadata(roomId: string): Promise { - const archivedRoomMetadataPath = `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/.room_metadata/${roomId}/room_metadata.json`; - - try { - await this.s3Service.deleteObjects([archivedRoomMetadataPath]); - this.logger.verbose(`Archived room metadata deleted for room ${roomId} in recordings bucket`); - } catch (error) { - this.logger.error( - `Error deleting archived room metadata for room ${roomId} in recordings bucket: ${error}` - ); - this.handleError(error, `Error deleting archived room metadata for room ${roomId}`); - throw error; - } - } - - async getRecordingMedia( - recordingPath: string, - range?: { - end: number; - start: number; - } - ): Promise { - try { - this.logger.debug(`Retrieving recording media from S3 at path: ${recordingPath}`); - return await this.s3Service.getObjectAsStream(recordingPath, range); - } catch (error) { - this.handleError(error, `Error fetching recording media for path ${recordingPath}`); - throw error; - } - } - - /** - * Deletes multiple recording binary files from S3 storage using their file paths. - * - * @param recordingPaths - Array of file paths/keys identifying the recording files to delete from S3 - * @returns A Promise that resolves when all files have been successfully deleted - * @throws Will throw an error if the S3 delete operation fails - */ - async deleteRecordingBinaryFilesByPaths(recordingPaths: string[]): Promise { - try { - await this.s3Service.deleteObjects(recordingPaths); - this.logger.verbose(`Deleted recording binary files: ${recordingPaths.join(', ')}`); - } catch (error) { - this.handleError(error, `Error deleting recording binary files: ${recordingPaths.join(', ')}`); - throw error; - } - } - - async getRecordingMetadata(recordingId: string): Promise<{ recordingInfo: MRec; metadataFilePath: string }> { - try { - const metadataPath = RecordingHelper.buildMetadataFilePath(recordingId); - this.logger.debug(`Retrieving metadata for recording ${recordingId} from ${metadataPath}`); - const recordingInfo = (await this.s3Service.getObjectAsJson(metadataPath)) as MRec; - - if (!recordingInfo) { - throw errorRecordingNotFound(recordingId); - } - - this.logger.verbose(`Retrieved metadata for recording ${recordingId} from ${metadataPath}`); - - return { recordingInfo, metadataFilePath: metadataPath }; - } catch (error) { - this.handleError(error, `Error fetching recording metadata for recording ${recordingId}`); - throw error; - } - } - - /** - * Retrieves recording metadata from S3 storage by the specified path. - * - * @param recordingPath - The S3 path where the recording metadata is stored - * @returns A promise that resolves to the recording metadata object - * @throws Will throw an error if the S3 object retrieval fails or if the path is invalid - */ - async getRecordingMetadataByPath(recordingPath: string): Promise { - try { - return (await this.s3Service.getObjectAsJson(recordingPath)) as MRec; - } catch (error) { - this.handleError(error, `Error fetching recording metadata for path ${recordingPath}`); - throw error; - } - } - - async saveRecordingMetadata(recordingInfo: MRec): Promise { - try { - const metadataPath = RecordingHelper.buildMetadataFilePath(recordingInfo.recordingId); - await this.s3Service.saveObject(metadataPath, recordingInfo); - return recordingInfo; - } catch (error) { - this.handleError(error, `Error saving recording metadata for recording ${recordingInfo.recordingId}`); - throw error; - } - } - - /** - * Deletes multiple recording metadata files from S3 storage based on their file paths. - * - * @param metadataPaths - Array of file paths pointing to the metadata files to be deleted - * @returns A promise that resolves when all metadata files have been successfully deleted - * @throws May throw an error if any of the deletion operations fail - */ - async deleteRecordingMetadataByPaths(metadataPaths: string[]): Promise { - try { - await this.s3Service.deleteObjects(metadataPaths); - this.logger.verbose(`Deleted multiple recording metadata files: ${metadataPaths.join(', ')}`); - } catch (error) { - this.handleError(error, `Error deleting multiple recording metadata files: ${metadataPaths.join(', ')}`); - throw error; - } - } - - async getUser(username: string): Promise { - try { - const userKey = RedisKeyName.USER + username; - const user: MUser | null = await this.getFromRedis(userKey); - - if (!user) { - this.logger.debug(`User ${username} not found in Redis. Fetching from S3...`); - const s3Path = `${INTERNAL_CONFIG.S3_USERS_PREFIX}/${username}.json`; - return await this.getFromS3(s3Path); - } - - this.logger.debug(`User ${username} retrieved from Redis`); - return user; - } catch (error) { - this.handleError(error, `Error fetching user ${username}`); - return null; - } - } - - /** - * Saves a user object to both S3 storage and Redis cache atomically. - * - * This method attempts to persist the user data in S3 (as a JSON file) and in Redis (as a serialized string). - * If both operations succeed, the user is considered saved and the method returns the user object. - * If either operation fails, the method attempts to roll back any successful operation to maintain consistency. - * In case of failure, the error is logged and rethrown after rollback attempts. - * - * @param user - The user object to be saved. - * @returns A promise that resolves to the saved user object if both operations succeed. - * @throws An error if either the S3 or Redis operation fails, after attempting rollback. - */ - async saveUser(user: MUser): Promise { - const userKey = RedisKeyName.USER + user.username; - const redisPayload = JSON.stringify(user); - const s3Path = `${INTERNAL_CONFIG.S3_USERS_PREFIX}/${user.username}.json`; - - const [s3Result, redisResult] = await Promise.allSettled([ - this.s3Service.saveObject(s3Path, user), - this.redisService.set(userKey, redisPayload, false) - ]); - - if (s3Result.status === 'fulfilled' && redisResult.status === 'fulfilled') { - this.logger.info(`User ${user.username} saved successfully`); - return user; - } - - // Rollback any changes made by the successful operation - if (s3Result.status === 'fulfilled') { - try { - await this.s3Service.deleteObjects([s3Path]); - } catch (rollbackError) { - this.logger.error(`Error rolling back S3 save for user ${user.username}: ${rollbackError}`); - } - } - - if (redisResult.status === 'fulfilled') { - try { - await this.redisService.delete(userKey); - } catch (rollbackError) { - this.logger.error(`Error rolling back Redis set for user ${user.username}: ${rollbackError}`); - } - } - - // Return the error that occurred first - const failedOperation: PromiseRejectedResult = - s3Result.status === 'rejected' ? s3Result : (redisResult as PromiseRejectedResult); - const error = failedOperation.reason; - this.handleError(error, `Error saving user ${user.username}`); - throw error; - } - - /** - * Retrieves an object of type U from Redis by the given key. - * Returns null if the key is not found or an error occurs. - * - * @param key - The Redis key to fetch. - * @returns A promise that resolves to an object of type U or null. - */ - protected async getFromRedis(key: string): Promise { - try { - const response = await this.redisService.get(key); - - if (response) { - return JSON.parse(response) as U; - } - - return null; - } catch (error) { - this.logger.error(`Error fetching from Redis for key ${key}: ${error}`); - return null; - } - } - - /** - * Retrieves an object of type U from S3 at the specified path. - * Returns null if the object is not found. - * - * @param path - The S3 key or path to fetch. - * @returns A promise that resolves to an object of type U or null. - */ - protected async getFromS3(path: string): Promise { - try { - const response = await this.s3Service.getObjectAsJson(path); - - if (response) { - this.logger.verbose(`Object found in S3 at path: ${path}`); - return response as U; - } - - return null; - } catch (error) { - this.logger.error(`Error fetching from S3 for path ${path}: ${error}`); - return null; - } - } - - /** - * Extracts the room ID from the given S3 file path. - * Assumes the room ID is the directory name immediately preceding the file name. - * Example: 'path/to/roomId/file.json' -> 'roomId' - * - * @param filePath - The S3 object key representing the file path. - * @returns The extracted room ID or null if extraction fails. - */ - protected extractRoomId(filePath?: string): string | null { - if (!filePath) return null; - - const parts = filePath.split('/'); - const roomId = parts.slice(-2, -1)[0]; - - if (!roomId) { - this.logger.warn(`Invalid room file path: ${filePath}`); - return null; - } - - return roomId; - } - - /** - * Performs rollback of saved room data. - * - * @param roomId - The room identifier. - * @param s3Result - The result of the S3 save operation. - * @param redisResult - The result of the Redis set operation. - * @param s3Path - The S3 key used to save the room data. - * @param redisKey - The Redis key used to cache the room data. - */ - protected async rollbackRoomSave( - roomId: string, - s3Result: PromiseSettledResult, - redisResult: PromiseSettledResult, - s3Path: string, - redisKey: string - ): Promise { - if (s3Result.status === 'fulfilled') { - try { - await this.s3Service.deleteObjects([s3Path]); - } catch (rollbackError) { - this.logger.error(`Error rolling back S3 save for room ${roomId}: ${rollbackError}`); - } - } - - if (redisResult.status === 'fulfilled') { - try { - await this.redisService.delete(redisKey); - } catch (rollbackError) { - this.logger.error(`Error rolling back Redis set for room ${roomId}: ${rollbackError}`); - } - } - } - - protected handleError(error: unknown, message: string) { - if (error instanceof OpenViduMeetError) { - this.logger.error(`${message}: ${error.message}`); - } else { - this.logger.error(`${message}: Unexpected error`); - } - } -} diff --git a/backend/src/services/storage/providers/s3/s3-storage-key.builder.ts b/backend/src/services/storage/providers/s3/s3-storage-key.builder.ts new file mode 100644 index 0000000..e9ca54c --- /dev/null +++ b/backend/src/services/storage/providers/s3/s3-storage-key.builder.ts @@ -0,0 +1,41 @@ +import INTERNAL_CONFIG from '../../../../config/internal-config.js'; +import { RecordingHelper } from '../../../../helpers/recording.helper.js'; +import { StorageKeyBuilder } from '../../storage.interface.js'; + +export class S3KeyBuilder implements StorageKeyBuilder { + buildGlobalPreferencesKey(): string { + return `global-preferences.json`; + } + + buildMeetRoomKey(roomId: string): string { + return `${INTERNAL_CONFIG.S3_ROOMS_PREFIX}/${roomId}/${roomId}.json`; + } + + buildAllMeetRoomsKey(): string { + return `${INTERNAL_CONFIG.S3_ROOMS_PREFIX}`; + } + + buildArchivedMeetRoomKey(roomId: string): string { + return `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/.room_metadata/${roomId}/room_metadata.json`; + } + + buildMeetRecordingKey(recordingId: string): string { + const { roomId, egressId, uid } = RecordingHelper.extractInfoFromRecordingId(recordingId); + + return `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/.metadata/${roomId}/${egressId}/${uid}.json`; + } + + buildBinaryRecordingKey(recordingId: string): string { + const { roomId, uid } = RecordingHelper.extractInfoFromRecordingId(recordingId); + return `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/${roomId}/${roomId}--${uid}.mp4`; + } + + buildAllMeetRecordingsKey(roomId?: string): string { + const roomSegment = roomId ? `/${roomId}` : ''; + return `${INTERNAL_CONFIG.S3_RECORDINGS_PREFIX}/.metadata${roomSegment}`; + } + + buildUserKey(userId: string): string { + return `${INTERNAL_CONFIG.S3_USERS_PREFIX}/${userId}.json`; + } +} diff --git a/backend/src/services/storage/providers/s3/s3-storage.provider.ts b/backend/src/services/storage/providers/s3/s3-storage.provider.ts new file mode 100644 index 0000000..aecf2c2 --- /dev/null +++ b/backend/src/services/storage/providers/s3/s3-storage.provider.ts @@ -0,0 +1,140 @@ +import { inject, injectable } from 'inversify'; +import { Readable } from 'stream'; +import { LoggerService, S3Service } from '../../../index.js'; +import { StorageProvider } from '../../storage.interface.js'; + +/** + * Basic S3 storage provider that implements only primitive storage operations. + */ +@injectable() +export class S3StorageProvider implements StorageProvider { + constructor( + @inject(LoggerService) protected logger: LoggerService, + @inject(S3Service) protected s3Service: S3Service + ) {} + + /** + * Retrieves an object from S3 as a JSON object. + */ + async getObject>(key: string): Promise { + try { + this.logger.debug(`Getting object from S3: ${key}`); + const result = await this.s3Service.getObjectAsJson(key); + return result as T; + } catch (error) { + this.logger.debug(`Object not found in S3: ${key}`); + return null; + } + } + + /** + * Stores an object in S3 as JSON. + */ + async putObject>(key: string, data: T): Promise { + try { + this.logger.debug(`Storing object in S3: ${key}`); + await this.s3Service.saveObject(key, data as Record); + this.logger.verbose(`Successfully stored object in S3: ${key}`); + } catch (error) { + this.logger.error(`Error storing object in S3 ${key}: ${error}`); + throw error; + } + } + + /** + * Deletes a single object from S3. + */ + async deleteObject(key: string): Promise { + try { + this.logger.debug(`Deleting object from S3: ${key}`); + await this.s3Service.deleteObjects([key]); + this.logger.verbose(`Successfully deleted object from S3: ${key}`); + } catch (error) { + this.logger.error(`Error deleting object from S3 ${key}: ${error}`); + throw error; + } + } + + /** + * Deletes multiple objects from S3. + */ + async deleteObjects(keys: string[]): Promise { + try { + this.logger.debug(`Deleting ${keys.length} objects from S3`); + await this.s3Service.deleteObjects(keys); + this.logger.verbose(`Successfully deleted ${keys.length} objects from S3`); + } catch (error) { + this.logger.error(`Error deleting objects from S3: ${error}`); + throw error; + } + } + + /** + * Checks if an object exists in S3. + */ + async exists(key: string): Promise { + try { + this.logger.debug(`Checking if object exists in S3: ${key}`); + return await this.s3Service.exists(key); + } catch (error) { + this.logger.debug(`Error checking object existence in S3 ${key}: ${error}`); + return false; + } + } + + /** + * Lists objects in S3 with a given prefix. + */ + async listObjects( + prefix: string, + maxItems?: number, + continuationToken?: string + ): Promise<{ + Contents?: Array<{ + Key?: string; + LastModified?: Date; + Size?: number; + ETag?: string; + }>; + IsTruncated?: boolean; + NextContinuationToken?: string; + }> { + try { + this.logger.debug(`Listing objects in S3 with prefix: ${prefix}`); + return await this.s3Service.listObjectsPaginated(prefix, maxItems, continuationToken); + } catch (error) { + this.logger.error(`Error listing objects in S3 with prefix ${prefix}: ${error}`); + throw error; + } + } + + /** + * Retrieves metadata headers for an object in S3. + */ + async getObjectHeaders(key: string): Promise<{ contentLength?: number; contentType?: string }> { + try { + this.logger.debug(`Getting object headers from S3: ${key}`); + const data = await this.s3Service.getHeaderObject(key); + return { + contentLength: data.ContentLength, + contentType: data.ContentType + }; + } catch (error) { + this.logger.error(`Error fetching object headers from S3 ${key}: ${error}`); + throw error; + } + } + + /** + * Retrieves an object from S3 as a readable stream. + */ + async getObjectAsStream(key: string, range?: { start: number; end: number }): Promise { + try { + this.logger.debug(`Getting object stream from S3: ${key}`); + return await this.s3Service.getObjectAsStream(key, range); + } catch (error) { + this.logger.error(`Error fetching object stream from S3 ${key}: ${error}`); + throw error; + } + } +} diff --git a/backend/src/services/s3.service.ts b/backend/src/services/storage/providers/s3/s3.service.ts similarity index 94% rename from backend/src/services/s3.service.ts rename to backend/src/services/storage/providers/s3/s3.service.ts index bfe79fb..d73a3de 100644 --- a/backend/src/services/s3.service.ts +++ b/backend/src/services/storage/providers/s3/s3.service.ts @@ -22,10 +22,10 @@ import { MEET_S3_SERVICE_ENDPOINT, MEET_S3_SUBBUCKET, MEET_S3_WITH_PATH_STYLE_ACCESS -} from '../environment.js'; -import { errorS3NotAvailable, internalError } from '../models/error.model.js'; -import { LoggerService } from './index.js'; -import INTERNAL_CONFIG from '../config/internal-config.js'; +} from '../../../../environment.js'; +import { errorS3NotAvailable, internalError } from '../../../../models/error.model.js'; +import { LoggerService } from '../../../index.js'; +import INTERNAL_CONFIG from '../../../../config/internal-config.js'; @injectable() export class S3Service { @@ -64,7 +64,11 @@ export class S3Service { * Saves an object to a S3 bucket. * Uses an internal retry mechanism in case of errors. */ - async saveObject(name: string, body: any, bucket: string = MEET_S3_BUCKET): Promise { + async saveObject( + name: string, + body: Record, + bucket: string = MEET_S3_BUCKET + ): Promise { const fullKey = this.getFullKey(name); try { @@ -76,10 +80,10 @@ export class S3Service { const result = await this.retryOperation(() => this.run(command)); this.logger.verbose(`S3: successfully saved object '${fullKey}' in bucket '${bucket}'`); return result; - } catch (error: any) { + } catch (error: unknown) { this.logger.error(`S3: error saving object '${fullKey}' in bucket '${bucket}': ${error}`); - if (error.code === 'ECONNREFUSED') { + if (error && typeof error === 'object' && 'code' in error && error.code === 'ECONNREFUSED') { throw errorS3NotAvailable(error); } diff --git a/backend/src/services/storage/storage.factory.ts b/backend/src/services/storage/storage.factory.ts index afa9804..d78c1b1 100644 --- a/backend/src/services/storage/storage.factory.ts +++ b/backend/src/services/storage/storage.factory.ts @@ -1,30 +1,33 @@ import { inject, injectable } from 'inversify'; -import { MEET_PREFERENCES_STORAGE_MODE } from '../../environment.js'; -import { LoggerService, S3StorageProvider, StorageProvider } from '../index.js'; +import { LoggerService } from '../index.js'; +import { StorageKeyBuilder, StorageProvider } from './storage.interface.js'; +import { container, STORAGE_TYPES } from '../../config/dependency-injector.config.js'; /** - * Factory class responsible for creating the appropriate storage provider based on configuration. + * Factory class responsible for creating the appropriate basic storage provider + * based on configuration. * - * This factory determines which storage implementation to use based on the `MEET_PREFERENCES_STORAGE_MODE` - * environment variable. Currently supports S3 storage, with more providers potentially added in the future. + * This factory determines which basic storage implementation to use based on the + * `MEET_PREFERENCES_STORAGE_MODE` environment variable. It creates providers that + * handle only basic CRUD operations, following the Single Responsibility Principle. + * + * Domain-specific logic should be handled in the MeetStorageService layer. */ @injectable() export class StorageFactory { - constructor( - @inject(S3StorageProvider) protected s3StorageProvider: S3StorageProvider, - @inject(LoggerService) protected logger: LoggerService - ) {} + constructor(@inject(LoggerService) protected logger: LoggerService) {} - create(): StorageProvider { - const storageMode = MEET_PREFERENCES_STORAGE_MODE; - - switch (storageMode) { - case 's3': - return this.s3StorageProvider; - - default: - this.logger.info('No preferences storage mode specified. Defaulting to S3.'); - return this.s3StorageProvider; - } + /** + * Creates a basic storage provider based on the configured storage mode. + * + * @returns StorageProvider instance configured for the specified storage backend + */ + create(): { provider: StorageProvider; keyBuilder: StorageKeyBuilder } { + // The actual binding is handled in the DI configuration + // This factory just returns the pre-configured instances + return { + provider: container.get(STORAGE_TYPES.StorageProvider), + keyBuilder: container.get(STORAGE_TYPES.KeyBuilder) + }; } } diff --git a/backend/src/services/storage/storage.interface.ts b/backend/src/services/storage/storage.interface.ts index 64546fb..001c81f 100644 --- a/backend/src/services/storage/storage.interface.ts +++ b/backend/src/services/storage/storage.interface.ts @@ -1,53 +1,67 @@ -import { GlobalPreferences, MeetRecordingInfo, MeetRoom, User } from '@typings-ce'; import { Readable } from 'stream'; /** - * An interface that defines the contract for storage providers in the OpenVidu Meet application. - * Storage providers handle persistence of global application preferences, rooms, recordings metadata and users. + * Basic storage interface that defines primitive storage operations. + * This interface follows the Single Responsibility Principle by focusing + * only on basic CRUD operations for object storage. * - * @template GPrefs - The type of global preferences, extending GlobalPreferences - * @template MRoom - The type of room data, extending MeetRoom - * @template MRec - The type of recording metadata, extending MeetRecordingInfo - * @template MUser - The type of user data, extending User - * - * Implementations of this interface should handle the persistent storage - * of application settings, room information, recording metadata, and user data, - * which could be backed by various storage solutions (database, file system, cloud storage, etc.). + * This allows easy integration of different storage backends (S3, PostgreSQL, + * FileSystem, etc.) without mixing domain-specific business logic. */ -export interface StorageProvider< - GPrefs extends GlobalPreferences = GlobalPreferences, - MRoom extends MeetRoom = MeetRoom, - MRec extends MeetRecordingInfo = MeetRecordingInfo, - MUser extends User = User -> { +export interface StorageProvider { /** - * Initializes the storage with default preferences if they are not already set. + * Retrieves an object from storage as a JSON object. * - * @param defaultPreferences - The default preferences to initialize with. - * @returns A promise that resolves when the initialization is complete. + * @param key - The storage key/path of the object + * @returns A promise that resolves to the parsed JSON object, or null if not found */ - initialize(defaultPreferences: GPrefs): Promise; + getObject>(key: string): Promise; /** - * Retrives the headers of an object stored in the storage provider. - * This is useful to get the content length and content type of the object without downloading it. + * Stores an object in storage as JSON. * - * @param filePath - The path of the file to retrieve headers for. + * @param key - The storage key/path where the object should be stored + * @param data - The object to store (will be serialized to JSON) + * @returns A promise that resolves when the object is successfully stored */ - getObjectHeaders(filePath: string): Promise<{ contentLength?: number; contentType?: string }>; + putObject>(key: string, data: T): Promise; /** - * Lists objects in the storage with optional pagination support. + * Deletes a single object from storage. * - * @param prefix - The prefix to filter objects by (acts as a folder path) + * @param key - The storage key/path of the object to delete + * @returns A promise that resolves when the object is successfully deleted + */ + deleteObject(key: string): Promise; + + /** + * Deletes multiple objects from storage. + * + * @param keys - Array of storage keys/paths of the objects to delete + * @returns A promise that resolves when all objects are successfully deleted + */ + deleteObjects(keys: string[]): Promise; + + /** + * Checks if an object exists in storage. + * + * @param key - The storage key/path to check + * @returns A promise that resolves to true if the object exists, false otherwise + */ + exists(key: string): Promise; + + /** + * Lists objects in storage with a given prefix (acts like a folder). + * + * @param prefix - The prefix to filter objects by * @param maxItems - Maximum number of items to return (optional) - * @param nextPageToken - Token for pagination to get the next page (optional) - * @returns Promise resolving to paginated list of objects with metadata + * @param continuationToken - Token for pagination (optional) + * @returns A promise that resolves to a paginated list of objects */ listObjects( prefix: string, maxItems?: number, - nextPageToken?: string + continuationToken?: string ): Promise<{ Contents?: Array<{ Key?: string; @@ -60,165 +74,80 @@ export interface StorageProvider< }>; /** - * Retrieves the global preferences of Openvidu Meet. + * Retrieves metadata headers for an object without downloading the content. * - * @returns A promise that resolves to the global preferences, or null if not set. + * @param key - The storage key/path of the object + * @returns A promise that resolves to object metadata */ - getGlobalPreferences(): Promise; - - /** - * Saves the given preferences. - * - * @param preferences - The preferences to save. - * @returns A promise that resolves to the saved preferences. - */ - saveGlobalPreferences(preferences: GPrefs): Promise; - - /** - * - * Retrieves the OpenVidu Meet Rooms. - * - * @param maxItems - The maximum number of items to retrieve. If not provided, all items will be retrieved. - * @param nextPageToken - The token for the next page of results. If not provided, the first page will be retrieved. - * @returns A promise that resolves to an object containing: - * - the retrieved rooms. - * - a boolean indicating if there are more items to retrieve. - * - an optional next page token. - */ - getMeetRooms( - maxItems?: number, - nextPageToken?: string - ): Promise<{ - rooms: MRoom[]; - isTruncated: boolean; - nextPageToken?: string; + getObjectHeaders(key: string): Promise<{ + contentLength?: number; + contentType?: string; }>; /** - * Retrieves the {@link MeetRoom}. + * Retrieves an object as a readable stream. + * Useful for large files or when you need streaming access. * - * @param roomId - The identifier of the room to retrieve. - * @returns A promise that resolves to the OpenVidu Room, or null if not found. - **/ - getMeetRoom(roomId: string): Promise; - - /** - * Saves the OpenVidu Meet Room. - * - * @param meetRoom - The OpenVidu Room to save. - * @returns A promise that resolves to the saved OpenVidu Room. - **/ - saveMeetRoom(meetRoom: MRoom): Promise; - - /** - * Deletes OpenVidu Meet Rooms. - * - * @param roomIds - The room IDs to delete. - * @returns A promise that resolves when the room have been deleted. - **/ - deleteMeetRooms(roomIds: string[]): Promise; - - /** - * Gets the archived metadata for a specific room. - * - * The archived metadata is necessary for checking the permissions of the recording viewer when the room is deleted. - * - * @param roomId - The name of the room to retrieve. + * @param key - The storage key/path of the object + * @param range - Optional byte range for partial content retrieval + * @returns A promise that resolves to a readable stream of the object content */ - getArchivedRoomMetadata(roomId: string): Promise | null>; - - /** - * Archives the metadata for a specific room. - * - * This is necessary for persisting the metadata of a room although it is deleted. - * The metadata will be used to check the permissions of the recording viewer. - * - * @param roomId: The room ID to archive. - */ - archiveRoomMetadata(roomId: string): Promise; - - /** - * Updates the archived metadata for a specific room. - * - * This is necessary for keeping the metadata of a room up to date. - * - * @param roomId: The room ID to update. - */ - updateArchivedRoomMetadata(roomId: string): Promise; - - /** - * Deletes the archived metadata for a specific room. - * - * @param roomId - The room ID to delete the archived metadata for. - */ - deleteArchivedRoomMetadata(roomId: string): Promise; - - /** - * Saves the recording metadata. - * - * @param recordingInfo - The recording information to save. - * @returns A promise that resolves to the saved recording information. - */ - saveRecordingMetadata(recordingInfo: MRec): Promise; - - /** - * Retrieves the recording metadata for a specific recording ID. - * - * @param recordingId - The unique identifier of the recording. - * @returns A promise that resolves to the recording metadata, or null if not found. - */ - getRecordingMetadata(recordingId: string): Promise<{ recordingInfo: MRec; metadataFilePath: string }>; - - /** - * Retrieves the recording metadata for multiple recording IDs. - * - * @param recordingPath - The path of the recording file to retrieve metadata for. - * @returns A promise that resolves to the recording metadata, or null if not found. - */ - getRecordingMetadataByPath(recordingPath: string): Promise; - - /** - * Deletes multiple recording metadata files by their paths. - * - * @param metadataPaths - An array of metadata file paths to delete. - */ - deleteRecordingMetadataByPaths(metadataPaths: string[]): Promise; - - /** - * Retrieves the media content of a recording file. - * - * @param recordingPath - The path of the recording file to retrieve. - * @param range - An optional range object specifying the start and end byte positions to retrieve. - */ - getRecordingMedia( - recordingPath: string, - range?: { - end: number; - start: number; - } - ): Promise; - - /** - * Deletes multiple recording binary files by their paths. - * - * @param recordingPaths - An array of recording file paths to delete. - * @returns A promise that resolves when the recording binary files have been deleted. - */ - deleteRecordingBinaryFilesByPaths(recordingPaths: string[]): Promise; - - /** - * Retrieves the user data for a specific username. - * - * @param username - The username of the user to retrieve. - * @returns A promise that resolves to the user data, or null if not found. - */ - getUser(username: string): Promise; - - /** - * Saves the user data. - * - * @param user - The user data to save. - * @returns A promise that resolves to the saved user data. - */ - saveUser(user: MUser): Promise; + getObjectAsStream(key: string, range?: { start: number; end: number }): Promise; +} + +/** + * Interface for building storage keys used throughout the application. + * Provides methods to generate standardized keys for different types of data storage operations. + */ +export interface StorageKeyBuilder { + /** + * Builds the key for global preferences storage. + */ + buildGlobalPreferencesKey(): string; + /** + * Builds the key for a specific room. + * + * @param roomId - The unique identifier of the meeting room + */ + buildMeetRoomKey(roomId: string): string; + + /** + * Builds the key for all meeting rooms. + */ + buildAllMeetRoomsKey(): string; + + /** + * Builds the key for archived room metadata. + * + * @param roomId - The unique identifier of the meeting room + */ + buildArchivedMeetRoomKey(roomId: string): string; + + /** + * Builds the key for a specific recording. + * + * @param recordingId - The unique identifier of the recording + */ + buildBinaryRecordingKey(recordingId: string): string; + + /** + * Builds the key for a specific recording metadata. + * + * @param recordingId - The unique identifier of the recording + */ + buildMeetRecordingKey(recordingId: string): string; + + /** + * Builds the key for all recordings in a room or globally. + * + * @param roomId - Optional room identifier to filter recordings by room + */ + buildAllMeetRecordingsKey(roomId?: string): string; + + /** + * Builds the key for a specific user + * + * @param userId - The unique identifier of the user + */ + buildUserKey(userId: string): string; } diff --git a/backend/src/services/storage/storage.service.ts b/backend/src/services/storage/storage.service.ts index 0f16494..c990c9b 100644 --- a/backend/src/services/storage/storage.service.ts +++ b/backend/src/services/storage/storage.service.ts @@ -10,17 +10,31 @@ import { MEET_WEBHOOK_URL } from '../../environment.js'; import { MeetLock, PasswordHelper } from '../../helpers/index.js'; -import { errorRoomNotFound, internalError, OpenViduMeetError } from '../../models/error.model.js'; -import { LoggerService, MutexService, StorageFactory, StorageProvider } from '../index.js'; +import { + errorRecordingNotFound, + errorRecordingRangeNotSatisfiable, + errorRoomNotFound, + internalError, + OpenViduMeetError, + RedisKeyName +} from '../../models/index.js'; +import { LoggerService, MutexService, RedisService } from '../index.js'; +import { StorageFactory } from './storage.factory.js'; +import { StorageKeyBuilder, StorageProvider } from './storage.interface.js'; /** - * A service for managing storage operations related to OpenVidu Meet rooms and preferences. + * Domain-specific storage service for OpenVidu Meet. * - * This service provides an abstraction layer over the underlying storage implementation, - * handling initialization, retrieval, and persistence of global preferences and room data. + * This service handles all domain-specific logic for rooms, recordings, and preferences, + * while delegating basic storage operations to the StorageProvider. + * + * This architecture follows the Single Responsibility Principle: + * - StorageProvider: Handles only basic CRUD operations + * - MeetStorageService: Handles domain-specific business logic * * @template GPrefs - Type for global preferences, extends GlobalPreferences * @template MRoom - Type for room data, extends MeetRoom + * @template MRec - Type for recording data, extends MeetRecordingInfo */ @injectable() export class MeetStorageService< @@ -30,56 +44,28 @@ export class MeetStorageService< MUser extends User = User > { protected storageProvider: StorageProvider; + protected keyBuilder: StorageKeyBuilder; constructor( @inject(LoggerService) protected logger: LoggerService, @inject(StorageFactory) protected storageFactory: StorageFactory, - @inject(MutexService) protected mutexService: MutexService + @inject(MutexService) protected mutexService: MutexService, + @inject(RedisService) protected redisService: RedisService ) { - this.storageProvider = this.storageFactory.create(); + const { provider, keyBuilder } = this.storageFactory.create(); + this.storageProvider = provider; + this.keyBuilder = keyBuilder; } - async getObjectHeaders(filePath: string): Promise<{ contentLength?: number; contentType?: string }> { - try { - const headers = await this.storageProvider.getObjectHeaders(filePath); - this.logger.verbose(`Object headers retrieved: ${JSON.stringify(headers)}`); - return headers; - } catch (error) { - this.handleError(error, 'Error retrieving object headers'); - throw internalError('Getting object headers'); - } - } + // ========================================== + // GLOBAL PREFERENCES DOMAIN LOGIC + // ========================================== /** - * Lists objects in the storage with optional pagination support. - * - * @param prefix - The prefix to filter objects by (acts as a folder path) - * @param maxItems - Maximum number of items to return (optional) - * @param nextPageToken - Token for pagination to get the next page (optional) - * @returns Promise resolving to paginated list of objects with metadata - */ - listObjects( - prefix: string, - maxItems?: number, - nextPageToken?: string - ): Promise<{ - Contents?: Array<{ - Key?: string; - LastModified?: Date; - Size?: number; - ETag?: string; - }>; - IsTruncated?: boolean; - NextContinuationToken?: string; - }> { - return this.storageProvider.listObjects(prefix, maxItems, nextPageToken); - } - - /** - * Initializes default preferences if not already initialized and saves the admin user. + * Initializes default preferences if not already initialized. * @returns {Promise} Default global preferences. */ - async initialize(): Promise { + async initializeGlobalPreferences(): Promise { try { // Acquire a global lock to prevent multiple initializations at the same time when running in HA mode const lock = await this.mutexService.acquire(MeetLock.getGlobalPreferencesLock(), ms('30s')); @@ -92,9 +78,27 @@ export class MeetStorageService< } this.logger.verbose('Initializing global preferences with default values'); - const preferences = await this.getDefaultPreferences(); - await this.storageProvider.initialize(preferences); - + const redisKey = RedisKeyName.GLOBAL_PREFERENCES; + const storageKey = this.keyBuilder.buildGlobalPreferencesKey(); + + const preferences = this.buildDefaultPreferences(); + this.logger.verbose('Initializing global preferences with default values'); + const existing = await this.getFromCacheAndStorage(redisKey, storageKey); + + if (!existing) { + await this.saveCacheAndStorage(redisKey, storageKey, preferences); + this.logger.info('Global preferences initialized with default values'); + } else { + // Check if it's from a different project + const existingProjectId = (existing as GlobalPreferences)?.projectId; + const newProjectId = (preferences as GlobalPreferences)?.projectId; + + if (existingProjectId !== newProjectId) { + this.logger.info('Different project detected, overwriting global preferences'); + await this.saveCacheAndStorage(redisKey, storageKey, preferences); + } + } + // Save the default admin user const admin = { username: MEET_ADMIN_USER, @@ -104,59 +108,59 @@ export class MeetStorageService< await this.saveUser(admin); } catch (error) { this.handleError(error, 'Error initializing default preferences'); + throw internalError('Failed to initialize global preferences'); } } - /** - * Retrieves the global preferences, initializing them if necessary. - * @returns {Promise} - */ async getGlobalPreferences(): Promise { - let preferences = await this.storageProvider.getGlobalPreferences(); + const redisKey = RedisKeyName.GLOBAL_PREFERENCES; + const storageKey = this.keyBuilder.buildGlobalPreferencesKey(); - if (preferences) return preferences as GPrefs; + const preferences = await this.getFromCacheAndStorage(redisKey, storageKey); - await this.initialize(); - preferences = await this.storageProvider.getGlobalPreferences(); + if (preferences) return preferences; - if (!preferences) { - this.logger.error('Global preferences not found after initialization'); - throw internalError('getting global preferences'); - } + // Build and save default preferences if not found in cache or storage + await this.initializeGlobalPreferences(); - return preferences as GPrefs; + return this.buildDefaultPreferences(); } /** - * Saves the global preferences to the storage provider. - * @param {GPrefs} preferences - * @returns {Promise} + * Saves global preferences to the storage provider. + * @param {GPrefs} preferences - The global preferences to save. + * @returns {Promise} The saved global preferences. */ async saveGlobalPreferences(preferences: GPrefs): Promise { this.logger.info('Saving global preferences'); - return this.storageProvider.saveGlobalPreferences(preferences) as Promise; + const redisKey = RedisKeyName.GLOBAL_PREFERENCES; + const storageKey = this.keyBuilder.buildGlobalPreferencesKey(); + return await this.saveCacheAndStorage(redisKey, storageKey, preferences); } - /** - * Saves the meet room to the storage provider. - * - * @param meetRoom - The room object to be saved - * @returns A promise that resolves to the saved room object - */ + // ========================================== + // ROOM DOMAIN LOGIC + // ========================================== + async saveMeetRoom(meetRoom: MRoom): Promise { - this.logger.info(`Saving OpenVidu room ${meetRoom.roomId}`); - return this.storageProvider.saveMeetRoom(meetRoom) as Promise; + const { roomId } = meetRoom; + this.logger.info(`Saving OpenVidu room ${roomId}`); + const redisKey = RedisKeyName.ROOM + roomId; + const storageKey = this.keyBuilder.buildMeetRoomKey(roomId); + + return await this.saveCacheAndStorage(redisKey, storageKey, meetRoom); } /** - * Retrieves a paginated list of rooms from the storage provider. + * Retrieves a paginated list of meeting rooms from storage. * - * @param maxItems - Optional maximum number of rooms to retrieve in a single request - * @param nextPageToken - Optional token for pagination to get the next page of results - * @returns A promise that resolves to an object containing: - * - rooms: Array of MRoom objects representing the rooms + * @param maxItems - Optional maximum number of rooms to retrieve per page + * @param nextPageToken - Optional token for pagination to get the next set of results + * @returns Promise that resolves to an object containing: + * - rooms: Array of MRoom objects retrieved from storage * - isTruncated: Boolean indicating if there are more results available * - nextPageToken: Optional token for retrieving the next page of results + * @throws Error if the storage operation fails or encounters an unexpected error */ async getMeetRooms( maxItems?: number, @@ -166,160 +170,314 @@ export class MeetStorageService< isTruncated: boolean; nextPageToken?: string; }> { - return this.storageProvider.getMeetRooms(maxItems, nextPageToken) as Promise<{ - rooms: MRoom[]; - isTruncated: boolean; - nextPageToken?: string; - }>; + try { + const allRoomsKey = this.keyBuilder.buildAllMeetRoomsKey(); + const { Contents, IsTruncated, NextContinuationToken } = await this.storageProvider.listObjects( + allRoomsKey, + maxItems, + nextPageToken + ); + + const rooms: MRoom[] = []; + + if (Contents && Contents.length > 0) { + const roomPromises = Contents.map(async (item) => { + if (item.Key && item.Key.endsWith('.json')) { + try { + const room = await this.storageProvider.getObject(item.Key); + return room; + } catch (error) { + this.logger.warn(`Failed to load room from ${item.Key}: ${error}`); + return null; + } + } + + return null; + }); + + const roomResults = await Promise.all(roomPromises); + rooms.push(...roomResults.filter((room): room is Awaited => room !== null)); + } + + return { + rooms, + isTruncated: IsTruncated || false, + nextPageToken: NextContinuationToken + }; + } catch (error) { + this.handleError(error, 'Error retrieving rooms'); + throw error; + } + } + + async getMeetRoom(roomId: string): Promise { + const redisKey = RedisKeyName.ROOM + roomId; + const storageKey = this.keyBuilder.buildMeetRoomKey(roomId); + + return await this.getFromCacheAndStorage(redisKey, storageKey); + } + + async deleteMeetRooms(roomIds: string[]): Promise { + const roomKeys = roomIds.map((roomId) => this.keyBuilder.buildMeetRoomKey(roomId)); + const redisKeys = roomIds.map((roomId) => RedisKeyName.ROOM + roomId); + + await this.deleteFromCacheAndStorageBatch(redisKeys, roomKeys); + } + + // ========================================== + // ARCHIVED ROOM METADATA DOMAIN LOGIC + // ========================================== + + async getArchivedRoomMetadata(roomId: string): Promise | null> { + const redisKey = RedisKeyName.ARCHIVED_ROOM + roomId; + const storageKey = this.keyBuilder.buildArchivedMeetRoomKey(roomId); + + return await this.getFromCacheAndStorage>(redisKey, storageKey); } /** - * Retrieves the room by its unique identifier. + * Archives room metadata by storing essential room information in both cache and persistent storage. * - * @param roomId - The unique identifier for the room. - * @returns A promise that resolves to the room's preferences. - * @throws Error if the room preferences are not found. + * This method retrieves the room data, extracts key metadata (moderator/publisher URLs and + * recording preferences), and saves it to an archived location for future reference. + * + * If an archived metadata for the room already exists, it will be overwritten. + * + * @param roomId - The unique identifier of the room to archive + * @throws {Error} When the room with the specified ID is not found + * @returns A promise that resolves when the archiving operation completes successfully */ - async getMeetRoom(roomId: string): Promise { - const meetRoom = await this.storageProvider.getMeetRoom(roomId); + async archiveRoomMetadata(roomId: string): Promise { + const redisKey = RedisKeyName.ARCHIVED_ROOM + roomId; + const storageKey = this.keyBuilder.buildArchivedMeetRoomKey(roomId); - if (!meetRoom) { - this.logger.error(`Room not found for room ${roomId}`); + const room = await this.getMeetRoom(roomId); + + if (!room) { + this.logger.warn(`Room ${roomId} not found, cannot archive metadata`); throw errorRoomNotFound(roomId); } - return meetRoom as MRoom; + const archivedRoom: Partial = { + moderatorRoomUrl: room.moderatorRoomUrl, + publisherRoomUrl: room.publisherRoomUrl, + preferences: { + recordingPreferences: room.preferences?.recordingPreferences + } + } as Partial; + + await this.saveCacheAndStorage>(redisKey, storageKey, archivedRoom); } - /** - * Deletes multiple rooms from storage. - * - * @param roomIds - Array of room identifiers to be deleted - * @returns A promise that resolves when all rooms have been successfully deleted - * @throws May throw an error if the deletion operation fails for any of the rooms - */ - async deleteMeetRooms(roomIds: string[]): Promise { - return this.storageProvider.deleteMeetRooms(roomIds); - } - - /** - * Retrieves metadata for an archived room by its ID. - * - * @param roomId - The unique identifier of the room to retrieve metadata for - * @returns A promise that resolves to partial room metadata if found, or null if not found - */ - async getArchivedRoomMetadata(roomId: string): Promise | null> { - return this.storageProvider.getArchivedRoomMetadata(roomId) as Promise | null>; - } - - /** - * Archives the metadata for a specific room. - * - * @param roomId - The unique identifier of the room whose metadata should be archived - * @returns A Promise that resolves when the archival operation is complete - * @throws May throw an error if the archival operation fails or if the room ID is invalid - */ - async archiveRoomMetadata(roomId: string): Promise { - return this.storageProvider.archiveRoomMetadata(roomId); - } - - /** - * Updates the metadata of an archived room. - * - * @param roomId - The unique identifier of the room whose archived metadata should be updated - * @returns A promise that resolves when the archived room metadata has been successfully updated - * @throws May throw an error if the room ID is invalid or if the storage operation fails - */ - async updateArchivedRoomMetadata(roomId: string): Promise { - return this.storageProvider.updateArchivedRoomMetadata(roomId); - } - - /** - * Deletes the archived metadata for a specific room. - * - * @param roomId - The unique identifier of the room whose archived metadata should be deleted - * @returns A promise that resolves when the archived room metadata has been successfully deleted - * @throws May throw an error if the deletion operation fails or if the room ID is invalid - */ async deleteArchivedRoomMetadata(roomId: string): Promise { - return this.storageProvider.deleteArchivedRoomMetadata(roomId); + const redisKey = RedisKeyName.ARCHIVED_ROOM + roomId; + const storageKey = this.keyBuilder.buildArchivedMeetRoomKey(roomId); + + await this.deleteFromCacheAndStorage(redisKey, storageKey); + this.logger.verbose(`Archived room metadata deleted for room ${roomId} in recordings bucket`); } - /** - * Saves recording metadata to the storage provider. - * - * @param recordingInfo - The recording metadata object to be saved - * @returns A promise that resolves to the saved recording metadata object - */ + // ========================================== + // RECORDING DOMAIN LOGIC + // ========================================== + async saveRecordingMetadata(recordingInfo: MRec): Promise { - return this.storageProvider.saveRecordingMetadata(recordingInfo) as Promise; + const redisKey = RedisKeyName.RECORDING + recordingInfo.recordingId; + const storageKey = this.keyBuilder.buildMeetRecordingKey(recordingInfo.recordingId); + return await this.saveCacheAndStorage(redisKey, storageKey, recordingInfo); } /** - * Retrieves the metadata for a specific recording. + * Retrieves all recordings from storage, optionally filtered by room ID. * - * @param recordingId - The unique identifier of the recording - * @returns A promise that resolves to an object containing the recording information and metadata file path - * @throws May throw an error if the recording is not found or if there's an issue accessing the storage provider + * @param roomId - Optional room identifier to filter recordings. If not provided, retrieves all recordings. + * @param maxItems - Optional maximum number of items to return per page for pagination. + * @param nextPageToken - Optional token for pagination to retrieve the next page of results. + * + * @returns A promise that resolves to an object containing: + * - `recordings`: Array of recording metadata objects (MRec) + * - `isTruncated`: Optional boolean indicating if there are more results available + * - `nextContinuationToken`: Optional token to retrieve the next page of results + * + * @throws Will throw an error if storage retrieval fails or if there's an issue processing the recordings + * + * @remarks + * This method handles pagination and filters out any recordings that fail to load. + * Failed recordings are logged as warnings but don't cause the entire operation to fail. + * The method logs debug information about the retrieval process and summary statistics. */ - async getRecordingMetadata(recordingId: string): Promise<{ recordingInfo: MRec; metadataFilePath: string }> { - return this.storageProvider.getRecordingMetadata(recordingId) as Promise<{ - recordingInfo: MRec; - metadataFilePath: string; - }>; - } + async getAllRecordings( + roomId?: string, + maxItems?: number, + nextPageToken?: string + ): Promise<{ recordings: MRec[]; isTruncated?: boolean; nextContinuationToken?: string }> { + try { + const searchKey = this.keyBuilder.buildAllMeetRecordingsKey(roomId); + const scope = roomId ? ` for room ${roomId}` : ''; - /** - * Retrieves metadata for recordings by their file path. - * - * @param recordingPath - The path of the recording file to retrieve metadata for - * @returns A promise that resolves to - */ - async getRecordingMetadataByPath(recordingPath: string): Promise { - return this.storageProvider.getRecordingMetadataByPath(recordingPath) as Promise; - } + this.logger.debug(`Retrieving recordings${scope} with key: ${searchKey}`); + const { Contents, IsTruncated, NextContinuationToken } = await this.storageProvider.listObjects( + searchKey, + maxItems, + nextPageToken + ); - /** - * Retrieves recording media as a readable stream from the storage provider. - * - * @param recordingPath - The path to the recording file in storage - * @param range - Optional byte range for partial content retrieval - * @param range.start - Starting byte position - * @param range.end - Ending byte position - * @returns A Promise that resolves to a Readable stream of the recording media - */ - async getRecordingMedia( - recordingPath: string, - range?: { - end: number; - start: number; + if (!Contents || Contents.length === 0) { + this.logger.verbose(`No recordings found${scope}`); + return { recordings: [], isTruncated: false }; + } + + const metadataFiles = Contents; //Contents.filter((item) => item.Key && item.Key.endsWith('.json')); + + const recordingPromises = metadataFiles.map(async (item) => { + try { + const recording = await this.storageProvider.getObject(item.Key!); + return recording; + } catch (error) { + this.logger.warn(`Failed to load recording metadata from ${item.Key}: ${error}`); + return null; // Return null for failed loads, filter out later + } + }); + + // Wait for all recordings to load and filter out failures + const recordingResults = await Promise.all(recordingPromises); + const validRecordings = recordingResults.filter( + (recording): recording is Awaited => recording !== null && recording !== undefined + ); + + // Log results summary + const failedCount = recordingResults.length - validRecordings.length; + + if (failedCount > 0) { + this.logger.warn(`Failed to load ${failedCount} out of ${recordingResults.length} recordings${scope}`); + } + + this.logger.verbose(`Successfully retrieved ${validRecordings.length} recordings${scope}`); + + return { + recordings: validRecordings, + isTruncated: Boolean(IsTruncated), + nextContinuationToken: NextContinuationToken + }; + } catch (error) { + this.handleError(error, 'Error retrieving all recordings'); + throw error; + } + } + + async getRecordingMetadata(recordingId: string): Promise<{ recordingInfo: MRec; metadataFilePath: string }> { + try { + const redisKey = RedisKeyName.RECORDING + recordingId; + const storageKey = this.keyBuilder.buildMeetRecordingKey(recordingId); + + const recordingInfo = await this.getFromCacheAndStorage(redisKey, storageKey); + + if (!recordingInfo) { + throw errorRecordingNotFound(recordingId); + } + + this.logger.debug(`Retrieved recording for ${recordingId}`); + return { recordingInfo, metadataFilePath: storageKey }; + } catch (error) { + this.logger.error(`Error fetching recording metadata for recording ${recordingId}: ${error}`); + throw error; } - ): Promise { - return this.storageProvider.getRecordingMedia(recordingPath, range) as Promise; } /** - * Deletes multiple recording metadata files by their paths. + * Deletes a recording and its metadata by recordingId. + * This method handles the path building internally, making it agnostic to storage backend. * - * @param metadataPaths - Array of file paths to the recording metadata files to be deleted - * @returns A Promise that resolves when all metadata files have been successfully deleted - * @throws May throw an error if any of the deletion operations fail + * @param recordingId - The unique identifier of the recording to delete + * @returns Promise that resolves when both binary files and metadata are deleted */ - async deleteRecordingMetadataByPaths(metadataPaths: string[]): Promise { - return this.storageProvider.deleteRecordingMetadataByPaths(metadataPaths); + async deleteRecording(recordingId: string): Promise { + try { + const redisMetadataKey = RedisKeyName.RECORDING + recordingId; + const storageMetadataKey = this.keyBuilder.buildMeetRecordingKey(recordingId); + const binaryRecordingKey = this.keyBuilder.buildBinaryRecordingKey(recordingId); + + this.logger.info(`Deleting recording ${recordingId} with metadata key ${storageMetadataKey}`); + + // Delete both metadata and binary files + await Promise.all([ + this.deleteFromCacheAndStorage(redisMetadataKey, storageMetadataKey), + this.storageProvider.deleteObject(binaryRecordingKey) + ]); + + this.logger.verbose(`Successfully deleted recording ${recordingId}`); + } catch (error) { + this.handleError(error, `Error deleting recording ${recordingId}`); + throw error; + } } /** - * Deletes recording binary files from storage using the provided file paths. + * Deletes multiple recordings by recordingIds. * - * @param recordingPaths - Array of file paths pointing to the recording binary files to be deleted - * @returns A Promise that resolves when all specified recording files have been successfully deleted - * @throws May throw an error if any of the file deletion operations fail + * @param recordingIds - Array of recording identifiers to delete + * @returns Promise that resolves when all recordings are deleted */ - async deleteRecordingBinaryFilesByPaths(recordingPaths: string[]): Promise { - return this.storageProvider.deleteRecordingBinaryFilesByPaths(recordingPaths); + async deleteRecordings(recordingIds: string[]): Promise { + if (recordingIds.length === 0) { + this.logger.debug('No recordings to delete'); + return; + } + + try { + // Build all paths from recordingIds + const metadataKeys: string[] = []; + const redisKeys: string[] = []; + const binaryKeys: string[] = []; + + for (const recordingId of recordingIds) { + redisKeys.push(RedisKeyName.RECORDING + recordingId); + metadataKeys.push(this.keyBuilder.buildMeetRecordingKey(recordingId)); + binaryKeys.push(this.keyBuilder.buildBinaryRecordingKey(recordingId)); + } + + this.logger.debug(`Bulk deleting ${recordingIds.length} recordings`); + + // Delete all files in parallel using batch operations + await Promise.all([ + this.deleteFromCacheAndStorageBatch(redisKeys, metadataKeys), + this.storageProvider.deleteObjects(binaryKeys) + ]); + this.logger.verbose(`Successfully bulk deleted ${recordingIds.length} recordings`); + } catch (error) { + this.handleError(error, `Error deleting recordings: ${recordingIds.join(', ')}`); + throw error; + } } + async getRecordingMedia( + recordingId: string, + range?: { end: number; start: number } + ): Promise<{ fileSize: number | undefined; fileStream: Readable; start?: number; end?: number }> { + try { + const binaryRecordingKey = this.keyBuilder.buildBinaryRecordingKey(recordingId); + this.logger.debug(`Retrieving recording media for recording ${recordingId} from ${binaryRecordingKey}`); + + const fileSize = await this.getRecordingFileSize(binaryRecordingKey, recordingId); + const validatedRange = this.validateAndAdjustRange(range, fileSize, recordingId); + const fileStream = await this.storageProvider.getObjectAsStream(binaryRecordingKey, validatedRange); + + return { + fileSize, + fileStream, + start: validatedRange?.start, + end: validatedRange?.end + }; + } catch (error) { + this.logger.error(`Error fetching recording media for recording ${recordingId}: ${error}`); + throw error; + } + } + + // ========================================== + // USER DOMAIN LOGIC + // ========================================== + /** * Retrieves user data for a specific username. * @@ -327,7 +485,10 @@ export class MeetStorageService< * @returns A promise that resolves to the user data, or null if not found */ async getUser(username: string): Promise { - return this.storageProvider.getUser(username) as Promise; + const redisKey = RedisKeyName.USER + username; + const storageKey = this.keyBuilder.buildUserKey(username); + + return await this.getFromCacheAndStorage(redisKey, storageKey); } /** @@ -337,15 +498,246 @@ export class MeetStorageService< * @returns A promise that resolves to the saved user data */ async saveUser(user: MUser): Promise { - this.logger.info(`Saving user data for ${user.username}`); - return this.storageProvider.saveUser(user) as Promise; + const { username } = user; + const userRedisKey = RedisKeyName.USER + username; + const storageUserKey = this.keyBuilder.buildUserKey(username); + + return await this.saveCacheAndStorage(userRedisKey, storageUserKey, user); + } + + // ========================================== + // PRIVATE HELPER METHODS + // ========================================== + + // ========================================== + // HYBRID CACHE METHODS (Redis + Storage) + // ========================================== + + /** + * Saves data to both Redis cache and persistent storage with fallback handling. + * + * @param redisKey - The Redis key to store the data + * @param storageKey - The storage key/path for persistent storage + * @param data - The data to store + * @param redisTtl - Optional TTL for Redis cache (default: 1 hour) + * @returns Promise that resolves when data is saved to at least one location + */ + protected async saveCacheAndStorage(redisKey: string, storageKey: string, data: T): Promise { + const operations = [ + // Save to Redis (fast cache) + this.redisService.set(redisKey, JSON.stringify(data)).catch((error) => { + this.logger.warn(`Redis save failed for key ${redisKey}: ${error}`); + return Promise.reject({ type: 'redis', error }); + }), + + // Save to persistent storage + this.storageProvider.putObject(storageKey, data).catch((error) => { + this.logger.warn(`Storage save failed for key ${storageKey}: ${error}`); + return Promise.reject({ type: 'storage', error }); + }) + ]; + + try { + // Try to save to both locations + const results = await Promise.allSettled(operations); + + const redisResult = results[0]; + const storageResult = results[1]; + + // Check if at least one succeeded + const redisSuccess = redisResult.status === 'fulfilled'; + const storageSuccess = storageResult.status === 'fulfilled'; + + if (!redisSuccess && !storageSuccess) { + // Both failed - this is critical + const redisError = (redisResult as PromiseRejectedResult).reason; + const storageError = (storageResult as PromiseRejectedResult).reason; + + this.logger.error(`Save failed for both Redis and Storage:`, { + redisKey, + storageKey, + redisError: redisError.error, + storageError: storageError.error + }); + + throw new Error(`Failed to save data: Redis (${redisError.error}) and Storage (${storageError.error})`); + } + + // Log partial failures + if (!redisSuccess) { + const redisError = (redisResult as PromiseRejectedResult).reason; + this.logger.warn(`Redis save failed but storage succeeded for key ${redisKey}:`, redisError.error); + } + + if (!storageSuccess) { + const storageError = (storageResult as PromiseRejectedResult).reason; + this.logger.warn(`Storage save failed but Redis succeeded for key ${storageKey}:`, storageError.error); + } + + // Success if at least one location worked + this.logger.debug(`Save completed: Redis=${redisSuccess}, Storage=${storageSuccess}`); + return data; + } catch (error) { + this.handleError(error, `Error saving keys: ${redisKey}, ${storageKey}`); + throw error; + } + } + + /** + * Retrieves data from Redis cache first, falls back to storage if not found. + * Updates Redis cache if data is retrieved from storage. + * + * @param redisKey - The Redis key to check first + * @param storageKey - The storage key/path as fallback + * @returns Promise that resolves with the data or null if not found + */ + protected async getFromCacheAndStorage(redisKey: string, storageKey: string): Promise { + try { + // 1. Try Redis first (fast cache) + this.logger.debug(`Attempting to get data from Redis cache: ${redisKey}`); + + const cachedData = await this.redisService.get(redisKey); + + if (cachedData) { + this.logger.debug(`Cache HIT for key: ${redisKey}`); + + try { + return JSON.parse(cachedData) as T; + } catch (parseError) { + this.logger.warn(`Failed to parse cached data for key ${redisKey}: ${parseError}`); + // Continue to storage fallback + } + } else { + this.logger.debug(`Cache MISS for key: ${redisKey}`); + } + + // 2. Fallback to persistent storage + this.logger.debug(`Attempting to get data from storage: ${storageKey}`); + + const storageData = await this.storageProvider.getObject(storageKey); + + if (!storageData) { + this.logger.debug(`Data not found in storage for key: ${storageKey}`); + return null; + } + + // 3. Found in storage - update Redis cache for next time + this.logger.debug(`Storage HIT for key: ${storageKey}, updating cache`); + + try { + await this.redisService.set(redisKey, JSON.stringify(storageData)); + this.logger.debug(`Successfully updated cache for key: ${redisKey}`); + } catch (cacheUpdateError) { + // Cache update failure shouldn't affect the main operation + this.logger.warn(`Failed to update cache for key ${redisKey}: ${cacheUpdateError}`); + } + + return storageData; + } catch (error) { + this.handleError(error, `Error in hybrid cache get for keys: ${redisKey}, ${storageKey}`); + + throw error; // Re-throw unexpected errors + } + } + + /** + * Deletes data from both Redis cache and persistent storage. + * + * @param redisKey - The Redis key to delete + * @param storageKey - The storage key to delete + * @returns Promise that resolves when deletion is attempted on both locations + */ + protected async deleteFromCacheAndStorage(redisKey: string, storageKey: string): Promise { + return await this.deleteFromCacheAndStorageBatch([redisKey], [storageKey]); + } + + /** + * Deletes data from both Redis cache and persistent storage in batch. + * More efficient than multiple individual delete operations. + * + * @param deletions - Array of objects containing redisKey and storageKey pairs + * @returns Promise that resolves when batch deletion is attempted on both locations + */ + protected async deleteFromCacheAndStorageBatch(redisKeys: string[], storageKeys: string[]): Promise { + if (redisKeys.length === 0 && storageKeys.length === 0) { + this.logger.debug('No keys to delete in batch'); + return; + } + + this.logger.debug(`Batch deleting ${redisKeys.length} Redis keys and ${storageKeys.length} storage keys`); + + const operations = [ + // Batch delete from Redis (only if there are keys to delete) + redisKeys.length > 0 + ? this.redisService.delete(redisKeys).catch((error) => { + this.logger.warn(`Redis batch delete failed: ${error}`); + return Promise.reject({ type: 'redis', error, affectedKeys: redisKeys }); + }) + : Promise.resolve(0), + + // Batch delete from storage (only if there are keys to delete) + storageKeys.length > 0 + ? this.storageProvider.deleteObjects(storageKeys).catch((error) => { + this.logger.warn(`Storage batch delete failed: ${error}`); + return Promise.reject({ type: 'storage', error, affectedKeys: storageKeys }); + }) + : Promise.resolve() + ]; + + try { + const results = await Promise.allSettled(operations); + + const redisResult = results[0]; + const storageResult = results[1]; + + const redisSuccess = redisResult.status === 'fulfilled'; + const storageSuccess = storageResult.status === 'fulfilled'; + + if (redisKeys.length > 0) { + if (redisSuccess) { + const deletedCount = (redisResult as PromiseFulfilledResult).value; + this.logger.debug(`Redis batch delete succeeded: ${deletedCount} keys deleted`); + } else { + const redisError = (redisResult as PromiseRejectedResult).reason; + this.logger.warn(`Redis batch delete failed:`, redisError.error); + } + } + + if (storageKeys.length > 0) { + if (storageSuccess) { + this.logger.debug(`Storage batch delete succeeded: ${storageKeys.length} keys deleted`); + } else { + const storageError = (storageResult as PromiseRejectedResult).reason; + this.logger.warn(`Storage batch delete failed:`, storageError.error); + } + } + + this.logger.debug(`Batch delete completed: Redis=${redisSuccess}, Storage=${storageSuccess}`); + } catch (error) { + this.handleError(error, `Error in batch delete operation`); + throw error; + } + } + + /** + * Invalidates Redis cache for a specific key. + * Useful when you know data has changed and want to force next read from storage. + */ + protected async invalidateCache(redisKey: string): Promise { + try { + await this.redisService.delete(redisKey); + this.logger.debug(`Cache invalidated for key: ${redisKey}`); + } catch (error) { + this.logger.warn(`Failed to invalidate cache for key ${redisKey}: ${error}`); + // Don't throw - cache invalidation failure shouldn't break main flow + } } /** * Returns the default global preferences. * @returns {GPrefs} */ - protected async getDefaultPreferences(): Promise { + protected buildDefaultPreferences(): GPrefs { return { projectId: MEET_NAME_ID, webhooksPreferences: { @@ -363,16 +755,63 @@ export class MeetStorageService< } as GPrefs; } - /** - * Handles errors and logs them. - * @param {any} error - * @param {string} message - */ - protected handleError(error: OpenViduMeetError | unknown, message: string) { + protected async getRecordingFileSize(key: string, recordingId: string): Promise { + const { contentLength: fileSize } = await this.storageProvider.getObjectHeaders(key); + + if (!fileSize) { + this.logger.warn(`Recording media not found for recording ${recordingId}`); + throw errorRecordingNotFound(recordingId); + } + + return fileSize; + } + + protected validateAndAdjustRange( + range: { end: number; start: number } | undefined, + fileSize: number, + recordingId: string + ): { start: number; end: number } | undefined { + if (!range) return undefined; + + const { start, end: originalEnd } = range; + + // Validate input values + if (isNaN(start) || isNaN(originalEnd) || start < 0) { + this.logger.warn(`Invalid range values for recording ${recordingId}: start=${start}, end=${originalEnd}`); + this.logger.warn(`Returning full stream for recording ${recordingId}`); + return undefined; + } + + // Check if start is beyond file size + if (start >= fileSize) { + this.logger.error( + `Invalid range: start=${start} exceeds fileSize=${fileSize} for recording ${recordingId}` + ); + throw errorRecordingRangeNotSatisfiable(recordingId, fileSize); + } + + // Adjust end to not exceed file bounds + const adjustedEnd = Math.min(originalEnd, fileSize - 1); + + // Validate final range + if (start > adjustedEnd) { + this.logger.warn( + `Invalid range after adjustment: start=${start}, end=${adjustedEnd} for recording ${recordingId}` + ); + return undefined; + } + + this.logger.debug( + `Valid range for recording ${recordingId}: start=${start}, end=${adjustedEnd}, fileSize=${fileSize}` + ); + return { start, end: adjustedEnd }; + } + + protected handleError(error: unknown, context: string): void { if (error instanceof OpenViduMeetError) { - this.logger.error(`${message}: ${error.message}`); + this.logger.error(`${context}: ${error.message}`); } else { - this.logger.error(`${message}: Unexpected error`); + this.logger.error(`${context}: ${error}`); } } } diff --git a/backend/tests/integration/api/global-preferences/security.test.ts b/backend/tests/integration/api/global-preferences/security.test.ts index 77c8b68..f94d211 100644 --- a/backend/tests/integration/api/global-preferences/security.test.ts +++ b/backend/tests/integration/api/global-preferences/security.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeAll, describe, expect, it } from '@jest/globals'; +import { beforeAll, beforeEach, describe, expect, it } from '@jest/globals'; import { expectValidationError } from '../../../helpers/assertion-helpers.js'; import { getSecurityPreferences, @@ -6,6 +6,8 @@ import { updateSecurityPreferences } from '../../../helpers/request-helpers.js'; import { AuthMode, AuthType } from '../../../../src/typings/ce/index.js'; +import { container } from '../../../../src/config/dependency-injector.config.js'; +import { MeetStorageService } from '../../../../src/services/index.js'; const defaultPreferences = { authentication: { @@ -16,17 +18,18 @@ const defaultPreferences = { } }; -const restoreDefaultSecurityPreferences = async () => { - await updateSecurityPreferences(defaultPreferences); +const restoreDefaultGlobalPreferences = async () => { + const defaultPref = await container.get(MeetStorageService)['buildDefaultPreferences'](); + await container.get(MeetStorageService).saveGlobalPreferences(defaultPref); }; describe('Security Preferences API Tests', () => { - beforeAll(() => { + beforeAll(async () => { startTestServer(); }); - afterEach(async () => { - await restoreDefaultSecurityPreferences(); + beforeEach(async () => { + await restoreDefaultGlobalPreferences(); }); describe('Update security preferences', () => { diff --git a/backend/tests/integration/api/recordings/bulk-delete-recording.test.ts b/backend/tests/integration/api/recordings/bulk-delete-recording.test.ts index ef973b6..330118b 100644 --- a/backend/tests/integration/api/recordings/bulk-delete-recording.test.ts +++ b/backend/tests/integration/api/recordings/bulk-delete-recording.test.ts @@ -13,8 +13,9 @@ import { import { setupMultiRecordingsTestContext } from '../../../helpers/test-scenarios'; describe('Recording API Tests', () => { - beforeAll(() => { + beforeAll(async () => { startTestServer(); + await Promise.all([deleteAllRooms(), deleteAllRecordings()]); }); afterAll(async () => { diff --git a/backend/tests/integration/api/recordings/delete-recording.test.ts b/backend/tests/integration/api/recordings/delete-recording.test.ts index b652ad8..c06ee7b 100644 --- a/backend/tests/integration/api/recordings/delete-recording.test.ts +++ b/backend/tests/integration/api/recordings/delete-recording.test.ts @@ -15,8 +15,13 @@ import { import { setupMultiRecordingsTestContext } from '../../../helpers/test-scenarios'; describe('Recording API Tests', () => { - beforeAll(() => { + beforeAll(async () => { startTestServer(); + await Promise.all([deleteAllRooms(), deleteAllRecordings()]); + }); + + afterAll(async () => { + await Promise.all([deleteAllRooms(), deleteAllRecordings()]); }); describe('Delete Recording Tests', () => { diff --git a/backend/tests/integration/api/recordings/get-media-recording.test.ts b/backend/tests/integration/api/recordings/get-media-recording.test.ts index 4f4ef55..ffa0568 100644 --- a/backend/tests/integration/api/recordings/get-media-recording.test.ts +++ b/backend/tests/integration/api/recordings/get-media-recording.test.ts @@ -17,6 +17,8 @@ describe('Recording API Tests', () => { beforeAll(async () => { startTestServer(); + await Promise.all([deleteAllRooms(), deleteAllRecordings()]); + const testContext = await setupMultiRecordingsTestContext(1, 1, 1, '3s'); const roomData = testContext.getRoomByIndex(0)!; diff --git a/backend/tests/integration/api/recordings/get-recording.test.ts b/backend/tests/integration/api/recordings/get-recording.test.ts index c24244f..4ae4b1a 100644 --- a/backend/tests/integration/api/recordings/get-recording.test.ts +++ b/backend/tests/integration/api/recordings/get-recording.test.ts @@ -18,7 +18,8 @@ describe('Recording API Tests', () => { beforeAll(async () => { startTestServer(); - await deleteAllRecordings(); + await Promise.all([deleteAllRooms(), deleteAllRecordings()]); + // Create a room and join a participant context = await setupMultiRecordingsTestContext(1, 1, 1); ({ room, moderatorCookie, recordingId = '' } = context.getRoomByIndex(0)!); diff --git a/backend/tests/integration/api/recordings/get-recordings.test.ts b/backend/tests/integration/api/recordings/get-recordings.test.ts index d796a0e..230271b 100644 --- a/backend/tests/integration/api/recordings/get-recordings.test.ts +++ b/backend/tests/integration/api/recordings/get-recordings.test.ts @@ -1,4 +1,4 @@ -import { afterAll, afterEach, beforeAll, describe, expect, it } from '@jest/globals'; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from '@jest/globals'; import { MeetRecordingInfo, MeetRecordingStatus, MeetRoom } from '../../../../src/typings/ce/index.js'; import { expectSuccessListRecordingResponse, @@ -28,16 +28,14 @@ describe('Recordings API Tests', () => { beforeAll(async () => { startTestServer(); - await deleteAllRecordings(); }); describe('List Recordings Tests', () => { - afterEach(async () => { - await deleteAllRecordings(); + beforeEach(async () => { + await Promise.all([deleteAllRooms(), deleteAllRecordings()]); const response = await getAllRecordings(); expect(response.status).toBe(200); expectSuccessListRecordingResponse(response, 0, false, false); - }); afterAll(async () => { diff --git a/backend/tests/integration/api/recordings/race-conditions.test.ts b/backend/tests/integration/api/recordings/race-conditions.test.ts index df4735e..0be3d08 100644 --- a/backend/tests/integration/api/recordings/race-conditions.test.ts +++ b/backend/tests/integration/api/recordings/race-conditions.test.ts @@ -33,6 +33,8 @@ describe('Recording API Race Conditions Tests', () => { beforeAll(async () => { startTestServer(); + await Promise.all([deleteAllRooms(), deleteAllRecordings()]); + recordingService = container.get(RecordingService); }); diff --git a/backend/tests/integration/api/recordings/start-recording.test.ts b/backend/tests/integration/api/recordings/start-recording.test.ts index 623f49a..6eb0817 100644 --- a/backend/tests/integration/api/recordings/start-recording.test.ts +++ b/backend/tests/integration/api/recordings/start-recording.test.ts @@ -24,15 +24,15 @@ describe('Recording API Tests', () => { let context: TestContext | null = null; let room: MeetRoom, moderatorCookie: string; - beforeAll(() => { + beforeAll(async () => { startTestServer(); + await Promise.all([deleteAllRooms(), deleteAllRecordings()]); }); afterAll(async () => { await stopAllRecordings(moderatorCookie); await disconnectFakeParticipants(); - await deleteAllRooms(); - await deleteAllRecordings(); + await Promise.all([deleteAllRooms(), deleteAllRecordings()]); }); describe('Start Recording Tests', () => { diff --git a/backend/tests/integration/api/recordings/stop-recording.test.ts b/backend/tests/integration/api/recordings/stop-recording.test.ts index ef387a0..309c0e2 100644 --- a/backend/tests/integration/api/recordings/stop-recording.test.ts +++ b/backend/tests/integration/api/recordings/stop-recording.test.ts @@ -18,6 +18,8 @@ describe('Recording API Tests', () => { beforeAll(async () => { startTestServer(); + await Promise.all([deleteAllRooms(), deleteAllRecordings()]); + }); afterAll(async () => { diff --git a/backend/tests/integration/api/rooms/create-room.test.ts b/backend/tests/integration/api/rooms/create-room.test.ts index ff48942..b60a213 100644 --- a/backend/tests/integration/api/rooms/create-room.test.ts +++ b/backend/tests/integration/api/rooms/create-room.test.ts @@ -25,14 +25,14 @@ describe('Room API Tests', () => { }); describe('Room Creation Tests', () => { - it('โœ… Should create a room without autoDeletionDate (default behavior)', async () => { + it('Should create a room without autoDeletionDate (default behavior)', async () => { const room = await createRoom({ roomIdPrefix: ' Test Room ' }); expectValidRoom(room, 'TestRoom'); }); - it('โœ… Should create a room with a valid autoDeletionDate', async () => { + it('Should create a room with a valid autoDeletionDate', async () => { const room = await createRoom({ autoDeletionDate: validAutoDeletionDate, roomIdPrefix: ' .,-------}{ยก$#<+My Room *123 ' @@ -41,7 +41,7 @@ describe('Room API Tests', () => { expectValidRoom(room, 'MyRoom123', validAutoDeletionDate); }); - it('โœ… Should create a room when sending full valid payload', async () => { + it('Should create a room when sending full valid payload', async () => { const payload = { roomIdPrefix: ' =Example Room&/ ', autoDeletionDate: validAutoDeletionDate, diff --git a/typings/src/auth-preferences.ts b/typings/src/auth-preferences.ts index 84808e5..59a20d8 100644 --- a/typings/src/auth-preferences.ts +++ b/typings/src/auth-preferences.ts @@ -1,38 +1,38 @@ export interface AuthenticationPreferences { - authMethod: ValidAuthMethod; - authModeToAccessRoom: AuthMode; + authMethod: ValidAuthMethod; + authModeToAccessRoom: AuthMode; } /** * Authentication modes available to enter a room. */ export const enum AuthMode { - NONE = 'none', // No authentication required - MODERATORS_ONLY = 'moderators_only', // Only moderators need authentication - ALL_USERS = 'all_users' // All users need authentication + NONE = 'none', // No authentication required + MODERATORS_ONLY = 'moderators_only', // Only moderators need authentication + ALL_USERS = 'all_users', // All users need authentication } /** * Authentication method base interface. */ export interface AuthMethod { - type: AuthType; + type: AuthType; } /** * Enum for authentication types. */ export const enum AuthType { - SINGLE_USER = 'single-user' - // MULTI_USER = 'multi-user', - // OAUTH_ONLY = 'oauth-only' + SINGLE_USER = 'single-user', + // MULTI_USER = 'multi-user', + // OAUTH_ONLY = 'oauth-only' } /** * Authentication method: Single user with fixed credentials. */ export interface SingleUserAuth extends AuthMethod { - type: AuthType.SINGLE_USER; + type: AuthType.SINGLE_USER; } /** @@ -54,7 +54,8 @@ export interface SingleUserAuth extends AuthMethod { /** * Union type for allowed authentication methods. */ -export type ValidAuthMethod = SingleUserAuth /* | MultiUserAuth | OAuthOnlyAuth */; +export type ValidAuthMethod = + SingleUserAuth /* | MultiUserAuth | OAuthOnlyAuth */; /** * Configuration for OAuth authentication. diff --git a/typings/src/global-preferences.ts b/typings/src/global-preferences.ts index bce0ffb..80851a4 100644 --- a/typings/src/global-preferences.ts +++ b/typings/src/global-preferences.ts @@ -4,18 +4,18 @@ import { AuthenticationPreferences } from './auth-preferences.js'; * Represents global preferences for OpenVidu Meet. */ export interface GlobalPreferences { - projectId: string; - // roomFeaturesPreferences: RoomFeaturesPreferences; - webhooksPreferences: WebhookPreferences; - securityPreferences: SecurityPreferences; + projectId: string; + // roomFeaturesPreferences: RoomFeaturesPreferences; + webhooksPreferences: WebhookPreferences; + securityPreferences: SecurityPreferences; } export interface WebhookPreferences { - enabled: boolean; - url?: string; - // events: WebhookEvent[]; + enabled: boolean; + url?: string; + // events: WebhookEvent[]; } export interface SecurityPreferences { - authentication: AuthenticationPreferences; + authentication: AuthenticationPreferences; } diff --git a/typings/src/participant.ts b/typings/src/participant.ts index 0eda55c..5b0dee7 100644 --- a/typings/src/participant.ts +++ b/typings/src/participant.ts @@ -5,34 +5,34 @@ import { OpenViduMeetPermissions } from './permissions/openvidu-permissions.js'; * Options for a participant to join a room. */ export interface ParticipantOptions { - /** - * The unique identifier for the room. - */ - roomId: string; + /** + * The unique identifier for the room. + */ + roomId: string; - /** - * The name of the participant. - */ - participantName: string; + /** + * The name of the participant. + */ + participantName: string; - /** - * A secret key for room access. - */ - secret: string; + /** + * A secret key for room access. + */ + secret: string; } /** * Represents the permissions for an individual participant. */ export interface ParticipantPermissions { - livekit: LiveKitPermissions; - openvidu: OpenViduMeetPermissions; + livekit: LiveKitPermissions; + openvidu: OpenViduMeetPermissions; } /** * Represents the role of a participant in a room. */ export const enum ParticipantRole { - MODERATOR = 'moderator', - PUBLISHER = 'publisher' + MODERATOR = 'moderator', + PUBLISHER = 'publisher', } diff --git a/typings/src/room-preferences.ts b/typings/src/room-preferences.ts index 9aca289..9bbcf9b 100644 --- a/typings/src/room-preferences.ts +++ b/typings/src/room-preferences.ts @@ -2,30 +2,30 @@ * Interface representing the preferences for a room. */ export interface MeetRoomPreferences { - chatPreferences: MeetChatPreferences; - recordingPreferences: MeetRecordingPreferences; - virtualBackgroundPreferences: MeetVirtualBackgroundPreferences; + chatPreferences: MeetChatPreferences; + recordingPreferences: MeetRecordingPreferences; + virtualBackgroundPreferences: MeetVirtualBackgroundPreferences; } /** * Interface representing the preferences for recording. */ export interface MeetRecordingPreferences { - enabled: boolean; - allowAccessTo?: MeetRecordingAccess; + enabled: boolean; + allowAccessTo?: MeetRecordingAccess; } export const enum MeetRecordingAccess { - ADMIN = 'admin', // Only admins can access the recording - ADMIN_MODERATOR = 'admin-moderator', // Admins and moderators can access - ADMIN_MODERATOR_PUBLISHER = 'admin-moderator-publisher', // Admins, moderators and publishers can access - PUBLIC = 'public' // Everyone can access + ADMIN = 'admin', // Only admins can access the recording + ADMIN_MODERATOR = 'admin-moderator', // Admins and moderators can access + ADMIN_MODERATOR_PUBLISHER = 'admin-moderator-publisher', // Admins, moderators and publishers can access + PUBLIC = 'public', // Everyone can access } export interface MeetChatPreferences { - enabled: boolean; + enabled: boolean; } export interface MeetVirtualBackgroundPreferences { - enabled: boolean; + enabled: boolean; } diff --git a/typings/src/user.ts b/typings/src/user.ts index c9486d5..0b80030 100644 --- a/typings/src/user.ts +++ b/typings/src/user.ts @@ -1,13 +1,13 @@ export interface User { - username: string; - passwordHash: string; - roles: UserRole[]; + username: string; + passwordHash: string; + roles: UserRole[]; } export const enum UserRole { - ADMIN = 'admin', - USER = 'user', - APP = 'app' + ADMIN = 'admin', + USER = 'user', + APP = 'app', } export type UserDTO = Omit;