radiola-backend/src/now-playing/icy-now-playing.service.ts

import { Injectable, Logger } from '@nestjs/common';
import { Interval } from '@nestjs/schedule';
import { PrismaService } from '../prisma/prisma.service';
import { NowPlayingService } from './now-playing.service';
import * as http from 'http';
import * as https from 'https';

/**
 * Сбор now-playing для не-Record станций (DFM и др.) через ICY-метаданные потока.
 * Станций много (сотни), поэтому за один тик опрашиваем окно и сдвигаем курсор —
 * за несколько минут проходим все по кругу. Обложку и зачёт в чарты/обогащение
 * берёт на себя NowPlayingService.ingest (обложка подтянется из нашей БД).
 */
@Injectable()
export class IcyNowPlayingService {
  private readonly logger = new Logger(IcyNowPlayingService.name);
  private cursor = 0;
  private readonly windowSize = 70;

  constructor(
    private readonly prisma: PrismaService,
    private readonly nowPlayingService: NowPlayingService,
  ) {}

  @Interval(60000)
  async pollIcyNowPlaying() {
    // ЕМГ (emgsound) и DFM/Крутой (genre=DFM) обрабатываются отдельными сервисами
    // через их API — исключаем из ICY, чтобы не тратить слоты впустую.
    const where = {
      recordStationId: null,
      isOnline: true,
      genre: {
        notIn: [
          'DFM',
          'MAXIMUM',
          'Love Radio',
          'Radio Monte Carlo',
          'Radio ROKS',
        ],
      },
      NOT: { streamUrl: { contains: 'emgsound.ru' } },
    };
    const total = await this.prisma.station.count({ where });
    if (total === 0) return;
    if (this.cursor >= total) this.cursor = 0;
    const offset = this.cursor;

    const stations = await this.prisma.station.findMany({
      where,
      orderBy: { stationId: 'asc' },
      skip: offset,
      take: this.windowSize,
    });
    this.cursor += this.windowSize;

    let successCount = 0;

    for (let i = 0; i < stations.length; i += 10) {
      const batch = stations.slice(i, i + 10);
      const results = await Promise.allSettled(
        batch.map(async (station) => {
          const track = await this.parseIcyMetadata(station.streamUrl);
          if (!track || !track.artist || !track.song) return null;

          await this.nowPlayingService.ingest({
            stationDbId: station.id,
            stationNumericId: station.stationId,
            artist: track.artist,
            song: track.song,
            coverUrl: null,
          });
          return track;
        }),
      );

      for (const result of results) {
        if (result.status === 'fulfilled' && result.value) successCount++;
      }
    }

    this.logger.log(
      `ICY poll: ${successCount}/${stations.length} updated (offset ${offset}/${total})`,
    );
  }

  private async parseIcyMetadata(
    url: string,
  ): Promise<{ artist: string; song: string } | null> {
    return new Promise((resolve, reject) => {
      const client = url.startsWith('https') ? https : http;
      const req = client.get(
        url,
        { headers: { 'Icy-MetaData': '1' }, timeout: 5000 },
        (res) => {
          const metaint = parseInt(
            (res.headers['icy-metaint'] as string) || '0',
          );
          if (!metaint) {
            req.destroy();
            resolve(null);
            return;
          }

          let audioBytesRead = 0;
          let metaLength = 0;
          let metaBytesRead = 0;
          let metaBuffer = Buffer.alloc(0);
          let state: 'audio' | 'meta-length' | 'meta' = 'audio';

          res.on('data', (chunk: Buffer) => {
            let offset = 0;
            while (offset < chunk.length) {
              if (state === 'audio') {
                const need = metaint - audioBytesRead;
                const available = chunk.length - offset;
                const take = Math.min(need, available);
                audioBytesRead += take;
                offset += take;
                if (audioBytesRead >= metaint) {
                  state = 'meta-length';
                }
              } else if (state === 'meta-length') {
                metaLength = chunk[offset] * 16;
                offset++;
                if (metaLength === 0) {
                  audioBytesRead = 0;
                  state = 'audio';
                } else {
                  metaBuffer = Buffer.alloc(0);
                  metaBytesRead = 0;
                  state = 'meta';
                }
              } else if (state === 'meta') {
                const need = metaLength - metaBytesRead;
                const available = chunk.length - offset;
                const take = Math.min(need, available);
                metaBuffer = Buffer.concat([
                  metaBuffer,
                  chunk.slice(offset, offset + take),
                ]);
                metaBytesRead += take;
                offset += take;
                if (metaBytesRead >= metaLength) {
                  const metaStr = metaBuffer
                    .toString('utf-8')
                    .replace(/\x00/g, '');
                  const match = metaStr.match(/StreamTitle='([^']+)'/);
                  req.destroy();
                  if (!match) {
                    resolve(null);
                    return;
                  }
                  const raw = match[1].trim();
                  // Некоторые потоки (101.ru и др.) шлют в StreamTitle JSON-статус,
                  // а не название трека — это не трек, отсекаем.
                  if (raw.startsWith('{') || raw.startsWith('[')) {
                    resolve(null);
                    return;
                  }
                  const parts = raw.split(' - ', 2);
                  if (parts.length < 2) {
                    resolve({ artist: raw, song: raw });
                  } else {
                    resolve({
                      artist: parts[0].trim(),
                      song: parts[1].trim(),
                    });
                  }
                  return;
                }
              }
            }
          });

          res.on('error', (err) => {
            req.destroy();
            reject(err);
          });
          res.on('end', () => resolve(null));
        },
      );

      req.on('error', (err) => reject(err));
      req.on('timeout', () => {
        req.destroy();
        reject(new Error('Timeout'));
      });
    });
  }
}