node.js - Enviar actualizaciones del progreso de la API periódicamente y guardar el progreso

En mi aplicación NestJS, estoy intentando trabajar en algo que podría tardar entre 10 y 20 minutos en completarse. Entonces, el usuario hace clic en un botón y se llama a una API que comienza a realizar sus procesos, los cuales tomarían esta cantidad de tiempo.

Para dar una mejor idea, piensa en ello de esta manera: Hay una aplicación web donde hay un proceso que toma algo de tiempo. Terminas mostrando al usuario actualizaciones de progreso en el frontend. Entonces, el usuario podría ver "Generando resumen de tu documento" y luego "Haciendo más trabajo relevante", etc.

Mientras esto sucede, me gustaría que el progreso también se guardara para que si el usuario se aleja, aún pueda volver y continuar desde donde se quedó el progreso (es decir, la interfaz de usuario se actualizaría). Y una vez que todo esté completo, avanzaría.

Quiero que cuando el usuario llame a esa API, no se obstaculice el proceso ni se bloquee. Pensé en usar Eventos Enviados por el Servidor (SSE) para esto, pero SSE se desconectaría si el usuario navega fuera. Entonces, ¿cómo debería hacerlo? En una API donde estoy generando una respuesta de OpenAI, estoy usando SSE para una sensación más receptiva (este es un fragmento reducido de mi código):

@Sse(':id/generate-answer')
async answerWithMagic(
    @Param('id') id,
) {
    const messages = this.aiService.resolvePrompt();
    const stream = new Subject();
    this.aiService
        .generateCompletion({ messages, listenTokens: true, isJsonParsable: false })
        .pipe(
            takeWhile((data) => {
                if (data.eventType === LLM_END_EVENT) {
                    stream.next({
                        eventType: data.eventType,
                        data: { token: data.token },
                    });
                    return false;
                } else {
                    return true;
                }
            }),
        )
        .subscribe((data) => {
            stream.next({ eventType: data.eventType, data: { token: data.token } });
        });

    return stream;
}

¿Cómo guardo el progreso aquí? Hacer llamadas repetidas a la base de datos no sería muy eficiente, así que pensé en usar Redis donde almaceno el progreso allí, pero no estoy seguro de qué dirección tomar aquí con esto.

Respuesta #1

Creo que tu enfoque está bien.

La idea es ejecutar la tarea larga en segundo plano y usar Redis para llevar un registro del progreso. Para el cálculo de la tarea puedes usar BullMQ (cola basada en Redis) que es directamente compatible con NestJS: https://docs.nestjs.com/techniques/queues .

Entonces, técnicamente, tienes que crear un RedisService que contendrá los métodos para establecer y obtener el progreso de la tarea:

@Injectable()
export class RedisService {
  private redis: Redis;

  constructor() {
    this.redis = new Redis({
      host: process.env.REDIS_HOST || 'localhost',
      port: parseInt(process.env.REDIS_PORT, 10) || 6379,
    });
  }

  async setProgress(taskId: string, progress: any) {
    await this.redis.set(`progress:${taskId}`, JSON.stringify(progress), 'EX', 3600); // TTL 1 hour
  }

  async getProgress(taskId: string) {
    const progress = await this.redis.get(`progress:${taskId}`);
    return progress ? JSON.parse(progress) : null;
  }
}

Luego, necesitas inyectar el RedisService en el servicio que maneja la tarea para almacenar los progresos.

Después de eso, solo tienes que crear otro endpoint para recuperar las actualizaciones de la tarea:

  @Post(':id/generate-answer')
  async startTask(@Param('id') taskId: string) {
    // Run task with BullMQ
  }

  @Get(':id/progress')
  async getProgress(@Param('id') taskId: string) {
    const progress = await this.redisService.getProgress(taskId);
    return progress ?? { progress: 0, message: 'Not started' };
  }

El Frontend puede consultar /tasks/:id/progress cada pocos segundos (o reconectar SSE si permanecen).