feat: cache serialization for streams (#4175)

* feat: cache serialization for streams

* fix: lint erros

---------

Co-authored-by: Ihar <[email protected]>

Author: ihar-tsykala

api-gateway/src/helpers/index.ts

@@ -15,4 +15,5 @@ export * from './wallet.js';
 export * from './decorators/index.js';
 export * from './interceptors/index.js';
 export * from './entity-owner.js';
-export * from './interceptors/utils/index.js';
+export * from './interceptors/utils/index.js';
+export * from './stream-to-buffer.js';

api-gateway/src/helpers/interceptors/cache.ts

@@ -1,11 +1,14 @@
-import { Injectable, NestInterceptor, ExecutionContext, CallHandler, HttpException, HttpStatus } from '@nestjs/common';
+import { Injectable, NestInterceptor, ExecutionContext, CallHandler, HttpException, HttpStatus, StreamableFile } from '@nestjs/common';
 
 import { Observable, of, switchMap, tap } from 'rxjs';
 
 //services
 import { CacheService } from '../cache-service.js';
 import { Users } from '../users.js';
 
+//helpers
+import { streamToBuffer } from '../index.js';
+
 //utils
 import { getCacheKey } from './utils/index.js';
 
@@ -48,7 +51,11 @@ export class CacheInterceptor implements NestInterceptor {
                 if (cachedResponse) {
                     let result = JSON.parse(cachedResponse);
 
-                    if (result.type === 'buffer') {
+                    if (result.type === 'StreamableFile') {
+                        const buffer = Buffer.from(result.data, 'base64');
+                        result = new StreamableFile(buffer);
+                    }
+                    else if (result.type === 'buffer') {
                         result = Buffer.from(result.data, 'base64');
                     } else  {
                         result = result.data;
@@ -74,7 +81,11 @@ export class CacheInterceptor implements NestInterceptor {
                             result = request.locals;
                         }
 
-                        if (Buffer.isBuffer(result)) {
+                        if (response instanceof StreamableFile) {
+                            const buffer = await streamToBuffer(response.getStream());
+                            result = { type: 'StreamableFile', data: buffer.toString('base64') };
+                        }
+                        else if (Buffer.isBuffer(result)) {
                             result = { type: 'buffer', data: result.toString('base64') };
                         } else if (typeof response === 'object') {
                             result = { type: 'json', data: result };

api-gateway/src/helpers/stream-to-buffer.ts

@@ -0,0 +1,10 @@
+import { Readable } from 'stream';
+
+export function streamToBuffer(stream: Readable): Promise<Buffer> {
+    return new Promise((resolve, reject) => {
+        const chunks: Buffer[] = [];
+        stream.on('data', chunk => chunks.push(Buffer.from(chunk)));
+        stream.on('error', reject);
+        stream.on('end', () => resolve(Buffer.concat(chunks)));
+    });
+}