Add GCS support:
commit 567b6698c6e3de6a5c603140a37d4aac9d7ffc9d
Author: juancarmore <juancar_more2@hotmail.com>
Date: Tue Sep 23 13:50:37 2025 +0200
backend: clean up GCS configuration and improve format in storage provider and service
commit a1b04b9d8a3d143b040eda5afc24e99e90599cc7
Author: Piwccle <sergiosergi11@hotmail.com>
Date: Mon Sep 22 15:38:10 2025 +0200
backend: add Google Cloud Storage (GCS) support
- Updated package.json to include @google-cloud/storage dependency.
- Enhanced dependency injector to support GCS as a storage option.
- Modified environment configuration to include GCS settings.
- Implemented GCSStorageProvider for basic storage operations (get, put, delete, list).
- Created GCSService to handle interactions with Google Cloud Storage.
- Added health check and metadata retrieval methods for GCS.
- Updated logging to accommodate GCS operations.
This commit is contained in:
parent
f83c1ac2e4
commit
80abeeb65e
877
backend/package-lock.json
generated
877
backend/package-lock.json
generated
File diff suppressed because it is too large
Load Diff
@ -53,6 +53,7 @@
|
|||||||
"dependencies": {
|
"dependencies": {
|
||||||
"@aws-sdk/client-s3": "3.846.0",
|
"@aws-sdk/client-s3": "3.846.0",
|
||||||
"@azure/storage-blob": "12.27.0",
|
"@azure/storage-blob": "12.27.0",
|
||||||
|
"@google-cloud/storage": "^7.17.1",
|
||||||
"@sesamecare-oss/redlock": "1.4.0",
|
"@sesamecare-oss/redlock": "1.4.0",
|
||||||
"archiver": "7.0.1",
|
"archiver": "7.0.1",
|
||||||
"bcrypt": "5.1.1",
|
"bcrypt": "5.1.1",
|
||||||
|
|||||||
@ -6,6 +6,8 @@ import {
|
|||||||
AuthService,
|
AuthService,
|
||||||
DistributedEventService,
|
DistributedEventService,
|
||||||
FrontendEventService,
|
FrontendEventService,
|
||||||
|
GCSService,
|
||||||
|
GCSStorageProvider,
|
||||||
LiveKitService,
|
LiveKitService,
|
||||||
LivekitWebhookService,
|
LivekitWebhookService,
|
||||||
LoggerService,
|
LoggerService,
|
||||||
@ -86,6 +88,12 @@ const configureStorage = (storageMode: string) => {
|
|||||||
container.bind(ABSService).toSelf().inSingletonScope();
|
container.bind(ABSService).toSelf().inSingletonScope();
|
||||||
container.bind(ABSStorageProvider).toSelf().inSingletonScope();
|
container.bind(ABSStorageProvider).toSelf().inSingletonScope();
|
||||||
break;
|
break;
|
||||||
|
case 'gcs':
|
||||||
|
container.bind<StorageProvider>(STORAGE_TYPES.StorageProvider).to(GCSStorageProvider).inSingletonScope();
|
||||||
|
container.bind<StorageKeyBuilder>(STORAGE_TYPES.KeyBuilder).to(S3KeyBuilder).inSingletonScope();
|
||||||
|
container.bind(GCSService).toSelf().inSingletonScope();
|
||||||
|
container.bind(GCSStorageProvider).toSelf().inSingletonScope();
|
||||||
|
break;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
@ -51,9 +51,9 @@ export const {
|
|||||||
LIVEKIT_API_KEY = 'devkey',
|
LIVEKIT_API_KEY = 'devkey',
|
||||||
LIVEKIT_API_SECRET = 'secret',
|
LIVEKIT_API_SECRET = 'secret',
|
||||||
|
|
||||||
MEET_BLOB_STORAGE_MODE = 's3', // Options: 's3', 'abs'
|
MEET_BLOB_STORAGE_MODE = 's3', // Options: 's3', 'abs', 'gcs'
|
||||||
|
|
||||||
// S3 configuration
|
// S3 or GCS configuration
|
||||||
MEET_S3_BUCKET = 'openvidu-appdata',
|
MEET_S3_BUCKET = 'openvidu-appdata',
|
||||||
MEET_S3_SUBBUCKET = 'openvidu-meet',
|
MEET_S3_SUBBUCKET = 'openvidu-meet',
|
||||||
MEET_S3_SERVICE_ENDPOINT = 'http://localhost:9000',
|
MEET_S3_SERVICE_ENDPOINT = 'http://localhost:9000',
|
||||||
@ -90,7 +90,7 @@ export const {
|
|||||||
* Gets the base URL without trailing slash
|
* Gets the base URL without trailing slash
|
||||||
*/
|
*/
|
||||||
export const getBaseUrl = (): string => {
|
export const getBaseUrl = (): string => {
|
||||||
return MEET_BASE_URL.endsWith('/') ? MEET_BASE_URL.slice(0, -1) : MEET_BASE_URL;
|
return MEET_BASE_URL.endsWith('/') ? MEET_BASE_URL.slice(0, -1) : MEET_BASE_URL;
|
||||||
};
|
};
|
||||||
|
|
||||||
export function checkModuleEnabled() {
|
export function checkModuleEnabled() {
|
||||||
@ -158,6 +158,11 @@ export const logEnvVars = () => {
|
|||||||
console.log('MEET AZURE ACCOUNT KEY:', credential('****' + MEET_AZURE_ACCOUNT_KEY.slice(-3)));
|
console.log('MEET AZURE ACCOUNT KEY:', credential('****' + MEET_AZURE_ACCOUNT_KEY.slice(-3)));
|
||||||
console.log('MEET AZURE CONTAINER NAME:', text(MEET_AZURE_CONTAINER_NAME));
|
console.log('MEET AZURE CONTAINER NAME:', text(MEET_AZURE_CONTAINER_NAME));
|
||||||
console.log('---------------------------------------------------------');
|
console.log('---------------------------------------------------------');
|
||||||
|
} else if (MEET_BLOB_STORAGE_MODE === 'gcs') {
|
||||||
|
console.log('GCS Configuration');
|
||||||
|
console.log('---------------------------------------------------------');
|
||||||
|
console.log('MEET GCS BUCKET:', text(MEET_S3_BUCKET));
|
||||||
|
console.log('---------------------------------------------------------');
|
||||||
}
|
}
|
||||||
|
|
||||||
console.log('Redis Configuration');
|
console.log('Redis Configuration');
|
||||||
|
|||||||
@ -7,3 +7,5 @@ export * from './providers/s3/s3-storage-key.builder.js';
|
|||||||
export * from './providers/s3/s3-storage.provider.js';
|
export * from './providers/s3/s3-storage.provider.js';
|
||||||
export * from './providers/abs/abs.service.js';
|
export * from './providers/abs/abs.service.js';
|
||||||
export * from './providers/abs/abs-storage.provider.js';
|
export * from './providers/abs/abs-storage.provider.js';
|
||||||
|
export * from './providers/gcp/gcs.service.js';
|
||||||
|
export * from './providers/gcp/gcs-storage.provider.js';
|
||||||
|
|||||||
@ -0,0 +1,172 @@
|
|||||||
|
import { inject, injectable } from 'inversify';
|
||||||
|
import { Readable } from 'stream';
|
||||||
|
import { GCSService, LoggerService } from '../../../index.js';
|
||||||
|
import { StorageProvider } from '../../storage.interface.js';
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Basic GCS storage provider that implements only primitive storage operations.
|
||||||
|
*/
|
||||||
|
@injectable()
|
||||||
|
export class GCSStorageProvider implements StorageProvider {
|
||||||
|
constructor(
|
||||||
|
@inject(LoggerService) protected logger: LoggerService,
|
||||||
|
@inject(GCSService) protected gcsService: GCSService
|
||||||
|
) {}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Retrieves an object from GCS Storage as a JSON object.
|
||||||
|
*/
|
||||||
|
async getObject<T = Record<string, unknown>>(key: string): Promise<T | null> {
|
||||||
|
try {
|
||||||
|
this.logger.debug(`Getting object from GCS Storage: ${key}`);
|
||||||
|
const result = await this.gcsService.getObjectAsJson(key);
|
||||||
|
return result as T;
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.debug(`Object not found in GCS Storage: ${key}`);
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Stores an object in GCS Storage as JSON.
|
||||||
|
*/
|
||||||
|
async putObject<T = Record<string, unknown>>(key: string, data: T): Promise<void> {
|
||||||
|
try {
|
||||||
|
this.logger.debug(`Storing object in GCS Storage: ${key}`);
|
||||||
|
await this.gcsService.saveObject(key, data as Record<string, unknown>);
|
||||||
|
this.logger.verbose(`Successfully stored object in GCS Storage: ${key}`);
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error(`Error storing object in GCS Storage ${key}: ${error}`);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Deletes a single object from GCS Storage.
|
||||||
|
*/
|
||||||
|
async deleteObject(key: string): Promise<void> {
|
||||||
|
try {
|
||||||
|
this.logger.debug(`Deleting object from GCS Storage: ${key}`);
|
||||||
|
await this.gcsService.deleteObjects([key]);
|
||||||
|
this.logger.verbose(`Successfully deleted object from GCS Storage: ${key}`);
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error(`Error deleting object from GCS Storage ${key}: ${error}`);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Deletes multiple objects from GCS Storage.
|
||||||
|
*/
|
||||||
|
async deleteObjects(keys: string[]): Promise<void> {
|
||||||
|
try {
|
||||||
|
this.logger.debug(`Deleting ${keys.length} objects from GCS Storage`);
|
||||||
|
await this.gcsService.deleteObjects(keys);
|
||||||
|
this.logger.verbose(`Successfully deleted ${keys.length} objects from GCS Storage`);
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error(`Error deleting objects from GCS Storage: ${error}`);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Checks if an object exists in GCS Storage.
|
||||||
|
*/
|
||||||
|
async exists(key: string): Promise<boolean> {
|
||||||
|
try {
|
||||||
|
this.logger.debug(`Checking if object exists in GCS Storage: ${key}`);
|
||||||
|
return await this.gcsService.exists(key);
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.debug(`Error checking object existence in GCS Storage ${key}: ${error}`);
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Lists objects in GCS Storage 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 GCS Storage with prefix: ${prefix}`);
|
||||||
|
const result = await this.gcsService.listObjectsPaginated(prefix, maxItems, continuationToken);
|
||||||
|
|
||||||
|
// Transform GCS response to match the expected interface
|
||||||
|
return {
|
||||||
|
Contents: result.Contents?.map((item) => ({
|
||||||
|
Key: item.Key,
|
||||||
|
LastModified: item.LastModified,
|
||||||
|
Size: item.Size,
|
||||||
|
ETag: undefined // GCS doesn't provide ETag in the same way as S3
|
||||||
|
})),
|
||||||
|
IsTruncated: !!result.NextContinuationToken,
|
||||||
|
NextContinuationToken: result.NextContinuationToken
|
||||||
|
};
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error(`Error listing objects in GCS Storage with prefix ${prefix}: ${error}`);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Retrieves metadata headers for an object in GCS Storage.
|
||||||
|
*/
|
||||||
|
async getObjectHeaders(key: string): Promise<{ contentLength?: number; contentType?: string }> {
|
||||||
|
try {
|
||||||
|
this.logger.debug(`Getting object headers from GCS Storage: ${key}`);
|
||||||
|
const data = await this.gcsService.getObjectHeaders(key);
|
||||||
|
return {
|
||||||
|
contentLength: data.ContentLength,
|
||||||
|
contentType: data.ContentType
|
||||||
|
};
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error(`Error fetching object headers from GCS Storage ${key}: ${error}`);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Retrieves an object from GCS Storage as a readable stream.
|
||||||
|
*/
|
||||||
|
async getObjectAsStream(key: string, range?: { start: number; end: number }): Promise<Readable> {
|
||||||
|
try {
|
||||||
|
this.logger.debug(`Getting object stream from GCS Storage: ${key}`);
|
||||||
|
return await this.gcsService.getObjectAsStream(key, range);
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error(`Error fetching object stream from GCS Storage ${key}: ${error}`);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Performs a health check on the GCS storage provider.
|
||||||
|
*/
|
||||||
|
async checkHealth(): Promise<{ accessible: boolean; bucketExists?: boolean; containerExists?: boolean }> {
|
||||||
|
try {
|
||||||
|
this.logger.debug('Performing GCS storage health check');
|
||||||
|
const healthResult = await this.gcsService.checkHealth();
|
||||||
|
return {
|
||||||
|
accessible: healthResult.accessible,
|
||||||
|
bucketExists: healthResult.bucketExists
|
||||||
|
};
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error(`GCS storage health check failed: ${error}`);
|
||||||
|
return {
|
||||||
|
accessible: false,
|
||||||
|
bucketExists: false
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
348
backend/src/services/storage/providers/gcp/gcs.service.ts
Normal file
348
backend/src/services/storage/providers/gcp/gcs.service.ts
Normal file
@ -0,0 +1,348 @@
|
|||||||
|
import { Bucket, File, GetFilesOptions, Storage } from '@google-cloud/storage';
|
||||||
|
import { inject, injectable } from 'inversify';
|
||||||
|
import { Readable } from 'stream';
|
||||||
|
import INTERNAL_CONFIG from '../../../../config/internal-config.js';
|
||||||
|
import { MEET_S3_BUCKET, MEET_S3_SUBBUCKET } from '../../../../environment.js';
|
||||||
|
import { errorS3NotAvailable, internalError } from '../../../../models/error.model.js';
|
||||||
|
import { LoggerService } from '../../../index.js';
|
||||||
|
|
||||||
|
@injectable()
|
||||||
|
export class GCSService {
|
||||||
|
protected storage: Storage;
|
||||||
|
protected bucket: Bucket;
|
||||||
|
|
||||||
|
constructor(@inject(LoggerService) protected logger: LoggerService) {
|
||||||
|
this.storage = new Storage();
|
||||||
|
this.bucket = this.storage.bucket(MEET_S3_BUCKET); // Use S3_BUCKET as GCS bucket name
|
||||||
|
this.logger.debug('GCS Storage Client initialized');
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Checks if a file exists in the specified GCS bucket.
|
||||||
|
*/
|
||||||
|
async exists(name: string, bucket: string = MEET_S3_BUCKET): Promise<boolean> {
|
||||||
|
try {
|
||||||
|
const bucketObj = bucket === MEET_S3_BUCKET ? this.bucket : this.storage.bucket(bucket);
|
||||||
|
const file = bucketObj.file(this.getFullKey(name));
|
||||||
|
const [exists] = await file.exists();
|
||||||
|
|
||||||
|
if (exists) {
|
||||||
|
this.logger.verbose(`GCS exists: file '${this.getFullKey(name)}' found in bucket '${bucket}'`);
|
||||||
|
} else {
|
||||||
|
this.logger.warn(`GCS exists: file '${this.getFullKey(name)}' not found in bucket '${bucket}'`);
|
||||||
|
}
|
||||||
|
|
||||||
|
return exists;
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.warn(
|
||||||
|
`GCS exists: error checking file '${this.getFullKey(name)}' in bucket '${bucket}': ${error}`
|
||||||
|
);
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Saves an object to a GCS bucket.
|
||||||
|
* Uses an internal retry mechanism in case of errors.
|
||||||
|
*/
|
||||||
|
async saveObject(name: string, body: Record<string, unknown>, bucket: string = MEET_S3_BUCKET): Promise<any> {
|
||||||
|
const fullKey = this.getFullKey(name);
|
||||||
|
|
||||||
|
try {
|
||||||
|
const bucketObj = bucket === MEET_S3_BUCKET ? this.bucket : this.storage.bucket(bucket);
|
||||||
|
const file = bucketObj.file(fullKey);
|
||||||
|
const result = await this.retryOperation(async () => {
|
||||||
|
await file.save(JSON.stringify(body), {
|
||||||
|
metadata: {
|
||||||
|
contentType: 'application/json'
|
||||||
|
}
|
||||||
|
});
|
||||||
|
return { success: true };
|
||||||
|
});
|
||||||
|
|
||||||
|
this.logger.verbose(`GCS saveObject: successfully saved object '${fullKey}' in bucket '${bucket}'`);
|
||||||
|
return result;
|
||||||
|
} catch (error: any) {
|
||||||
|
this.logger.error(`GCS saveObject: error saving object '${fullKey}' in bucket '${bucket}': ${error}`);
|
||||||
|
|
||||||
|
if (error.code === 'ECONNREFUSED' || error.code === 'ENOTFOUND') {
|
||||||
|
throw errorS3NotAvailable(error); // Reuse S3 error for compatibility
|
||||||
|
}
|
||||||
|
|
||||||
|
throw internalError('saving object to GCS Storage');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Bulk deletes objects from GCS Storage.
|
||||||
|
* @param keys Array of object keys to delete
|
||||||
|
* @param bucket GCS bucket name (default: MEET_S3_BUCKET)
|
||||||
|
*/
|
||||||
|
async deleteObjects(keys: string[], bucket: string = MEET_S3_BUCKET): Promise<any> {
|
||||||
|
try {
|
||||||
|
this.logger.verbose(
|
||||||
|
`GCS deleteObjects: attempting to delete ${keys.length} objects from bucket '${bucket}'`
|
||||||
|
);
|
||||||
|
|
||||||
|
const bucketObj = bucket === MEET_S3_BUCKET ? this.bucket : this.storage.bucket(bucket);
|
||||||
|
const deletePromises = keys.map((key) => {
|
||||||
|
const file = bucketObj.file(this.getFullKey(key));
|
||||||
|
return file.delete();
|
||||||
|
});
|
||||||
|
|
||||||
|
await Promise.all(deletePromises);
|
||||||
|
|
||||||
|
this.logger.verbose(`Successfully deleted objects: [${keys.join(', ')}]`);
|
||||||
|
this.logger.info(`Successfully deleted ${keys.length} objects from bucket '${bucket}'`);
|
||||||
|
return {
|
||||||
|
Deleted: keys.map((key) => ({ Key: this.getFullKey(key) })), // S3-like response format
|
||||||
|
Errors: []
|
||||||
|
};
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error(`GCS deleteObjects: error deleting objects in bucket '${bucket}': ${error}`);
|
||||||
|
throw internalError('deleting objects from GCS Storage');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* List objects with pagination.
|
||||||
|
*
|
||||||
|
* @param additionalPrefix Additional prefix relative to the subbucket.
|
||||||
|
* @param maxKeys Maximum number of objects to return. Defaults to 50.
|
||||||
|
* @param continuationToken Token to retrieve the next page.
|
||||||
|
* @param bucket Optional bucket name. Defaults to MEET_S3_BUCKET.
|
||||||
|
*
|
||||||
|
* @returns S3-compatible response object.
|
||||||
|
*/
|
||||||
|
async listObjectsPaginated(
|
||||||
|
additionalPrefix = '',
|
||||||
|
maxKeys = 50,
|
||||||
|
continuationToken?: string,
|
||||||
|
bucket: string = MEET_S3_BUCKET
|
||||||
|
): Promise<{
|
||||||
|
Contents?: Array<{ Key?: string; LastModified?: Date; Size?: number; ETag?: string }>;
|
||||||
|
NextContinuationToken?: string;
|
||||||
|
IsTruncated?: boolean;
|
||||||
|
KeyCount?: number;
|
||||||
|
}> {
|
||||||
|
const basePrefix = this.getFullKey(additionalPrefix);
|
||||||
|
this.logger.verbose(`GCS listObjectsPaginated: listing objects with prefix '${basePrefix}'`);
|
||||||
|
|
||||||
|
const options: GetFilesOptions = {
|
||||||
|
prefix: basePrefix,
|
||||||
|
maxResults: maxKeys
|
||||||
|
};
|
||||||
|
|
||||||
|
if (continuationToken && continuationToken !== 'undefined') {
|
||||||
|
options.pageToken = continuationToken;
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
const bucketObj = bucket === MEET_S3_BUCKET ? this.bucket : this.storage.bucket(bucket);
|
||||||
|
const [files, , response] = await bucketObj.getFiles(options);
|
||||||
|
|
||||||
|
interface GCSFileContent {
|
||||||
|
Key?: string;
|
||||||
|
LastModified?: Date;
|
||||||
|
Size?: number;
|
||||||
|
ETag?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
const contents: GCSFileContent[] = files.map(
|
||||||
|
(file: File): GCSFileContent => ({
|
||||||
|
Key: file.name,
|
||||||
|
LastModified: file.metadata.updated ? new Date(file.metadata.updated) : undefined,
|
||||||
|
Size: file.metadata.size ? parseInt(file.metadata.size as string) : undefined,
|
||||||
|
ETag: file.metadata.etag || undefined
|
||||||
|
})
|
||||||
|
);
|
||||||
|
|
||||||
|
const nextPageToken = (response as any)?.nextPageToken;
|
||||||
|
return {
|
||||||
|
Contents: contents,
|
||||||
|
NextContinuationToken: nextPageToken,
|
||||||
|
IsTruncated: !!nextPageToken,
|
||||||
|
KeyCount: contents.length
|
||||||
|
};
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error(`GCS listObjectsPaginated: error listing objects with prefix '${basePrefix}': ${error}`);
|
||||||
|
throw internalError('listing objects from GCS Storage');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async getObjectAsJson(name: string, bucket: string = MEET_S3_BUCKET): Promise<object | undefined> {
|
||||||
|
try {
|
||||||
|
const bucketObj = bucket === MEET_S3_BUCKET ? this.bucket : this.storage.bucket(bucket);
|
||||||
|
const file = bucketObj.file(this.getFullKey(name));
|
||||||
|
|
||||||
|
const [exists] = await file.exists();
|
||||||
|
|
||||||
|
if (!exists) {
|
||||||
|
this.logger.warn(`GCS getObjectAsJson: object '${name}' does not exist in bucket ${bucket}`);
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
const [content] = await file.download();
|
||||||
|
const parsed = JSON.parse(content.toString());
|
||||||
|
|
||||||
|
this.logger.verbose(
|
||||||
|
`GCS getObjectAsJson: successfully retrieved and parsed object ${name} from bucket ${bucket}`
|
||||||
|
);
|
||||||
|
return parsed;
|
||||||
|
} catch (error: any) {
|
||||||
|
if (error.code === 404) {
|
||||||
|
this.logger.warn(`GCS getObjectAsJson: object '${name}' does not exist in bucket ${bucket}`);
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (error.code === 'ECONNREFUSED' || error.code === 'ENOTFOUND') {
|
||||||
|
throw errorS3NotAvailable(error); // Reuse S3 error for compatibility
|
||||||
|
}
|
||||||
|
|
||||||
|
this.logger.error(
|
||||||
|
`GCS getObjectAsJson: error retrieving object '${name}' from bucket '${bucket}': ${error}`
|
||||||
|
);
|
||||||
|
throw internalError('getting object as JSON from GCS Storage');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async getObjectAsStream(
|
||||||
|
name: string,
|
||||||
|
range?: { start: number; end: number },
|
||||||
|
bucket: string = MEET_S3_BUCKET
|
||||||
|
): Promise<Readable> {
|
||||||
|
try {
|
||||||
|
const bucketObj = bucket === MEET_S3_BUCKET ? this.bucket : this.storage.bucket(bucket);
|
||||||
|
const file = bucketObj.file(this.getFullKey(name));
|
||||||
|
|
||||||
|
const options: any = {};
|
||||||
|
|
||||||
|
if (range) {
|
||||||
|
options.start = range.start;
|
||||||
|
options.end = range.end;
|
||||||
|
}
|
||||||
|
|
||||||
|
const stream = file.createReadStream(options);
|
||||||
|
|
||||||
|
this.logger.info(
|
||||||
|
`GCS getObjectAsStream: successfully retrieved object '${name}' as stream from bucket '${bucket}'`
|
||||||
|
);
|
||||||
|
return stream;
|
||||||
|
} catch (error: any) {
|
||||||
|
this.logger.error(
|
||||||
|
`GCS getObjectAsStream: error retrieving stream for object '${name}' from bucket '${bucket}': ${error}`
|
||||||
|
);
|
||||||
|
|
||||||
|
if (error.code === 'ECONNREFUSED' || error.code === 'ENOTFOUND') {
|
||||||
|
throw errorS3NotAvailable(error); // Reuse S3 error for compatibility
|
||||||
|
}
|
||||||
|
|
||||||
|
throw internalError('getting object as stream from GCS Storage');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async getObjectHeaders(name: string, bucket: string = MEET_S3_BUCKET): Promise<any> {
|
||||||
|
try {
|
||||||
|
const bucketObj = bucket === MEET_S3_BUCKET ? this.bucket : this.storage.bucket(bucket);
|
||||||
|
const file = bucketObj.file(this.getFullKey(name));
|
||||||
|
const [metadata] = await file.getMetadata();
|
||||||
|
|
||||||
|
this.logger.verbose(
|
||||||
|
`GCS getObjectHeaders: retrieved headers for object '${this.getFullKey(name)}' in bucket '${bucket}'`
|
||||||
|
);
|
||||||
|
|
||||||
|
// Return S3-compatible response format
|
||||||
|
return {
|
||||||
|
ContentLength: metadata.size,
|
||||||
|
LastModified: metadata.updated ? new Date(metadata.updated) : undefined,
|
||||||
|
ContentType: metadata.contentType,
|
||||||
|
ETag: metadata.etag,
|
||||||
|
Metadata: metadata.metadata || {}
|
||||||
|
};
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error(
|
||||||
|
`GCS getObjectHeaders: error retrieving headers for object '${this.getFullKey(name)}' in bucket '${bucket}': ${error}`
|
||||||
|
);
|
||||||
|
|
||||||
|
throw internalError('getting object headers from GCS Storage');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Health check for GCS Storage service and bucket accessibility.
|
||||||
|
* Verifies both service connectivity and bucket existence.
|
||||||
|
*/
|
||||||
|
async checkHealth(): Promise<{ accessible: boolean; bucketExists: boolean }> {
|
||||||
|
try {
|
||||||
|
// Check if we can access the bucket by getting its metadata
|
||||||
|
await this.bucket.getMetadata();
|
||||||
|
|
||||||
|
// If we reach here, both service and bucket are accessible
|
||||||
|
this.logger.verbose(`GCS health check: service accessible and bucket '${MEET_S3_BUCKET}' exists`);
|
||||||
|
return { accessible: true, bucketExists: true };
|
||||||
|
} catch (error: any) {
|
||||||
|
this.logger.error(`GCS health check failed: ${error.message}`);
|
||||||
|
|
||||||
|
// Check if it's a bucket-specific error
|
||||||
|
if (error.code === 404) {
|
||||||
|
this.logger.error(`GCS bucket '${MEET_S3_BUCKET}' does not exist`);
|
||||||
|
return { accessible: true, bucketExists: false };
|
||||||
|
}
|
||||||
|
|
||||||
|
// Service is not accessible
|
||||||
|
return { accessible: false, bucketExists: false };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Constructs the full key for a GCS Storage object by ensuring it includes the specified sub-bucket prefix.
|
||||||
|
* If the provided name already starts with the prefix, it is returned as-is.
|
||||||
|
* Otherwise, the prefix is prepended to the name.
|
||||||
|
*/
|
||||||
|
protected getFullKey(name: string): string {
|
||||||
|
const prefix = `${MEET_S3_SUBBUCKET}`; // Use S3_SUBBUCKET for compatibility
|
||||||
|
|
||||||
|
if (name.startsWith(prefix)) {
|
||||||
|
return name;
|
||||||
|
}
|
||||||
|
|
||||||
|
return `${prefix}/${name}`;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Retries a given asynchronous operation with exponential backoff.
|
||||||
|
*/
|
||||||
|
protected async retryOperation<T>(operation: () => Promise<T>): Promise<T> {
|
||||||
|
let attempt = 0;
|
||||||
|
let delayMs = Number(INTERNAL_CONFIG.S3_INITIAL_RETRY_DELAY_MS); // Reuse S3 config
|
||||||
|
const maxRetries = Number(INTERNAL_CONFIG.S3_MAX_RETRIES_ATTEMPTS_ON_SAVE_ERROR);
|
||||||
|
|
||||||
|
while (attempt < maxRetries) {
|
||||||
|
try {
|
||||||
|
this.logger.verbose(`GCS operation: attempt ${attempt + 1}`);
|
||||||
|
return await operation();
|
||||||
|
} catch (error) {
|
||||||
|
attempt++;
|
||||||
|
|
||||||
|
if (attempt >= maxRetries) {
|
||||||
|
this.logger.error(`GCS retryOperation: operation failed after ${maxRetries} attempts`);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.logger.warn(`GCS retryOperation: attempt ${attempt} failed. Retrying in ${delayMs}ms...`);
|
||||||
|
await this.sleep(delayMs);
|
||||||
|
// Exponential back off: delay increases by a factor of 2
|
||||||
|
delayMs *= 2;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
throw new Error('GCS retryOperation: exceeded maximum retry attempts without success');
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Internal helper to delay execution.
|
||||||
|
*/
|
||||||
|
protected sleep(ms: number): Promise<void> {
|
||||||
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
x
Reference in New Issue
Block a user