TTMusic/entry/src/main/ets/common/network/SmbStreamingHttpServer.ets

import Logger from '../util/Logger';
import { fileIo } from '@kit.CoreFileKit';
import { BusinessError } from '@kit.BasicServicesKit';
import { httpServer, HttpRequest, HttpResponse } from '@webabcd/harmony-httpserver';
import { readSmbRange, SmbConnectionInfo } from './SmbRangeReader';
import { readFtpRange, FtpConnectionInfo } from './FtpRangeReader';
import { connection } from '@kit.NetworkKit';

const TAG = 'SmbStreamingHttpServer';
const STREAM_ROUTE_PREFIX = '/smb-stream';
const DEFAULT_HOST = '127.0.0.1';
const MAX_CHUNK_SIZE = 512 * 1024; // 512KB per response
const RANGE_WAIT_TIMEOUT_MS = 60000;
const RANGE_POLL_INTERVAL_MS = 100;
const SESSION_TTL_MS = 10 * 60 * 1000;
const PORT_CANDIDATES: number[] = [18888, 18889, 18900, 18901, 18902];

type StreamingProtocol = 'smb' | 'ftp';

interface StreamingSessionOptions {
  sessionKey: string;
  cachePath: string;
  expectedSize?: number;
  mimeType?: string;
  fileName?: string;
  remotePath?: string;
  protocol?: StreamingProtocol;
  smbConnection?: SmbConnectionInfo;
  ftpConnection?: FtpConnectionInfo;
}

interface StreamingSession extends StreamingSessionOptions {
  lastAccess: number;
}

interface RangeInfo {
  start: number;
  end?: number;
  isRange: boolean;
}

interface ChunkResult {
  buffer: ArrayBuffer;
  start: number;
  end: number;
  total?: number;
}

class RangeNotSatisfiableError extends Error {
  availableRange: string;

  constructor(availableRange: string) {
    super('Requested Range Not Satisfiable');
    this.name = 'RangeNotSatisfiableError';
    this.availableRange = availableRange;
  }
}

export default class SmbStreamingHttpServer {
  private static instance?: SmbStreamingHttpServer;
  private port: number = -1;
  private sessions: Map<string, StreamingSession> = new Map();
  private startPromise?: Promise<void>;

  static getInstance(): SmbStreamingHttpServer {
    if (!SmbStreamingHttpServer.instance) {
      SmbStreamingHttpServer.instance = new SmbStreamingHttpServer();
    }
    return SmbStreamingHttpServer.instance;
  }

  async getStreamingUrl(options: StreamingSessionOptions): Promise<string> {
    await this.ensureServerStarted();
    const host = await this.resolveLocalHost();
    const now = Date.now();
    const existing = this.sessions.get(options.sessionKey);
    if (existing) {
      existing.lastAccess = now;
      existing.cachePath = options.cachePath;
      existing.expectedSize = options.expectedSize;
      existing.mimeType = options.mimeType;
      existing.fileName = options.fileName;
      existing.remotePath = options.remotePath;
      existing.protocol = options.protocol;
      existing.smbConnection = options.smbConnection;
      existing.ftpConnection = options.ftpConnection;
      Logger.debug(TAG, `更新SMB流会话: ${options.sessionKey}`);
    } else {
      this.sessions.set(options.sessionKey, {
        sessionKey: options.sessionKey,
        cachePath: options.cachePath,
        expectedSize: options.expectedSize,
        mimeType: options.mimeType,
        fileName: options.fileName,
        remotePath: options.remotePath,
        protocol: options.protocol,
        smbConnection: options.smbConnection,
        ftpConnection: options.ftpConnection,
        lastAccess: now
      });
      Logger.info(TAG, `注册SMB流会话: ${options.sessionKey}`);
    }
    this.cleanupExpiredSessions();
    return `http://${host}:${this.port}${STREAM_ROUTE_PREFIX}/${encodeURIComponent(options.sessionKey)}`;
  }

  private async resolveLocalHost(): Promise<string> {
    try {
      const netHandle = await connection.getDefaultNet();
      const addressGetter = (connection as unknown as { getAddressesByNetwork?: (handle: unknown) => Promise<unknown[]> })
        .getAddressesByNetwork;
      if (!addressGetter) {
        return DEFAULT_HOST;
      }
      const addresses = await addressGetter(netHandle);
      if (addresses && addresses.length > 0) {
        for (let i = 0; i < addresses.length; i++) {
          const raw = addresses[i] as Record<string, unknown>;
          const address = (raw.address ?? raw.addr ?? raw.ip) as string | undefined;
          if (!address) {
            continue;
          }
          if (address.startsWith('127.') || address === '0.0.0.0') {
            continue;
          }
          if (address.includes(':')) {
            continue;
          }
          return address;
        }
      }
    } catch (error) {
      Logger.warn(TAG, `解析本机IP失败: ${(error as Error).message}`);
    }
    return DEFAULT_HOST;
  }

  private async ensureServerStarted(): Promise<void> {
    if (this.port > 0) {
      return;
    }
    if (this.startPromise) {
      return this.startPromise;
    }
    this.startPromise = this.startServerInternal().finally(() => {
      this.startPromise = undefined;
    });
    return this.startPromise;
  }

  private async startServerInternal(): Promise<void> {
    httpServer.enableLog(false);
    this.port = await this.tryStartOnAvailablePort();
    httpServer.handleHttpRequestAsync((request: HttpRequest) => this.handleRequest(request));
    Logger.info(TAG, `SMB流HTTP服务启动，端口: ${this.port}`);
  }

  private async tryStartOnAvailablePort(): Promise<number> {
    for (let i = 0; i < PORT_CANDIDATES.length; i++) {
      const candidate = PORT_CANDIDATES[i];
      try {
        return await this.startOnPort(candidate);
      } catch (error) {
        const err = error as Error;
        Logger.warn(TAG, `端口${candidate}启动失败: ${err.message}`);
      }
    }
    throw new Error('无法启动SMB流HTTP服务，端口均不可用');
  }

  private startOnPort(port: number): Promise<number> {
    return new Promise((resolve, reject) => {
      try {
        httpServer.start(port, (error: BusinessError, realPort: number) => {
          if (error && error.code !== 0) {
            reject(new Error(`HTTP服务启动失败:${error.code}-${error.message}`));
            return;
          }
          resolve(realPort);
        });
      } catch (error) {
        reject(error as Error);
      }
    });
  }

  private async handleRequest(request: HttpRequest): Promise<HttpResponse> {
    try {
      this.cleanupExpiredSessions();
      if (!request || !request.url) {
        return this.buildPlainResponse(400, 'invalid request');
      }
      const sessionId = this.extractSessionId(request.url);
      Logger.debug(TAG, `收到请求 method=${request.method} url=${request.url} range=${this.getHeader(request.headers, 'range') ?? ''}`);
      if (!sessionId) {
        return this.buildPlainResponse(404, 'not found');
      }
      const session = this.sessions.get(sessionId);
      if (!session) {
        return this.buildPlainResponse(404, 'session expired');
      }
      session.lastAccess = Date.now();
      const method = (request.method || 'GET').toUpperCase();
      if (method !== 'GET' && method !== 'HEAD') {
        return this.buildPlainResponse(405, 'method not allowed');
      }
      if (method === 'HEAD') {
        return this.buildHeadResponse(session);
      }

      const rangeHeader = this.getHeader(request.headers, 'range');
      const rangeInfo = this.parseRangeHeader(rangeHeader);
      let chunk: ChunkResult;
      try {
        chunk = await this.readChunk(session, rangeInfo.start, rangeInfo.end);
      } catch (error) {
        const err = error as Error;
        if (err instanceof RangeNotSatisfiableError) {
          return this.buildRangeNotSatisfiableResponse(err.availableRange);
        }
        throw err;
      }
      return this.buildChunkResponse(session, chunk, rangeInfo.isRange);
    } catch (error) {
      const err = error as Error;
      Logger.error(TAG, `处理SMB流请求失败: ${err.message}`);
      return this.buildPlainResponse(500, err.message || 'internal error');
    }
  }

  private buildPlainResponse(statusCode: number, message: string): HttpResponse {
    return {
      statusCode,
      result: message,
      headers: {
        'Content-Type': 'text/plain; charset=utf-8'
      }
    } as HttpResponse;
  }

  private buildRangeNotSatisfiableResponse(availableRange: string): HttpResponse {
    return {
      statusCode: 416,
      result: 'Requested Range Not Satisfiable',
      headers: {
        'Content-Range': availableRange,
        'Content-Type': 'text/plain; charset=utf-8'
      }
    } as HttpResponse;
  }

  private buildHeadResponse(session: StreamingSession): HttpResponse {
    const headers: Record<string, string> = {
      'Accept-Ranges': 'bytes',
      'Content-Type': session.mimeType || 'application/octet-stream'
    };
    if (session.expectedSize && session.expectedSize > 0) {
      headers['Content-Length'] = session.expectedSize.toString();
    }
    return {
      statusCode: 200,
      headers,
      result: ''
    } as HttpResponse;
  }

  private buildChunkResponse(session: StreamingSession, chunk: ChunkResult, isRange: boolean): HttpResponse {
    const headers: Record<string, string> = {
      'Content-Type': session.mimeType || 'application/octet-stream',
      'Accept-Ranges': 'bytes',
      'Content-Length': (chunk.end - chunk.start + 1).toString()
    };
    if (chunk.total && chunk.total > 0) {
      headers['Content-Range'] = `bytes ${chunk.start}-${chunk.end}/${chunk.total}`;
    }
    return {
      statusCode: isRange ? 206 : 200,
      headers,
      result: chunk.buffer
    } as HttpResponse;
  }

  private extractSessionId(url: string): string | undefined {
    if (!url) {
      return undefined;
    }
    if (url.startsWith('http://') || url.startsWith('https://')) {
      const slashIndex = url.indexOf('/', url.indexOf('//') + 2);
      url = slashIndex >= 0 ? url.substring(slashIndex) : '/';
    }
    if (!url.startsWith(STREAM_ROUTE_PREFIX)) {
      return undefined;
    }
    let relative = url.substring(STREAM_ROUTE_PREFIX.length);
    if (relative.startsWith('/')) {
      relative = relative.substring(1);
    }
    const queryIndex = relative.indexOf('?');
    if (queryIndex >= 0) {
      relative = relative.substring(0, queryIndex);
    }
    if (!relative || relative.length === 0) {
      return undefined;
    }
    try {
      return decodeURIComponent(relative);
    } catch (error) {
      Logger.error(TAG, `sessionId解析失败: ${(error as Error).message}`);
      return undefined;
    }
  }

  private parseRangeHeader(header?: string): RangeInfo {
    if (!header || header.length === 0) {
      return { start: 0, isRange: false };
    }
    const match = header.match(/bytes=([0-9]*)-([0-9]*)/i);
    if (!match) {
      return { start: 0, isRange: false };
    }
    let start = match[1] ? parseInt(match[1]) : 0;
    const hasEnd = match[2] && match[2].length > 0;
    let end = hasEnd ? parseInt(match[2]) : undefined;
    if (isNaN(start) || start < 0) {
      start = 0;
    }
    if (end === undefined || isNaN(end) || end < start) {
      end = start + MAX_CHUNK_SIZE - 1;
    }
    const prefixRange = match[1] === undefined || match[1].length === 0;
    if (prefixRange) {
      const suffixBytes = parseInt(match[2]);
      if (!isNaN(suffixBytes) && suffixBytes > 0) {
        start = Math.max(0, end - suffixBytes + 1);
      }
    }
    if (end !== undefined && (isNaN(end) || end < start)) {
      end = start;
    }
    return { start, end, isRange: true };
  }

  private async readChunk(session: StreamingSession, requestedStart: number, requestedEnd?: number): Promise<ChunkResult> {
    const start = requestedStart >= 0 ? requestedStart : 0;
    let end = requestedEnd !== undefined ? requestedEnd : start + MAX_CHUNK_SIZE - 1;
    if (end - start + 1 > MAX_CHUNK_SIZE) {
      end = start + MAX_CHUNK_SIZE - 1;
    }

    const isFtp = session.protocol === 'ftp';
    const timeout = isFtp ? RANGE_WAIT_TIMEOUT_MS * 3 : RANGE_WAIT_TIMEOUT_MS;
    const pollInterval = isFtp ? RANGE_POLL_INTERVAL_MS * 2 : RANGE_POLL_INTERVAL_MS;

    const waitStart = Date.now();
    while (Date.now() - waitStart <= timeout) {
      const size = await this.tryGetFileSize(session.cachePath);
      Logger.debug(TAG, `range waiting start=${start} end=${end} currentSize=${size} cachePath=${session.cachePath} protocol=${session.protocol || 'unknown'}`);

      if (size >= 0 && size > start) {
        let fileEnd = Math.min(end, size - 1);
        if (session.expectedSize && session.expectedSize > 0) {
          fileEnd = Math.min(fileEnd, session.expectedSize - 1);
        }
        if (fileEnd >= start) {
          const length = fileEnd - start + 1;
          const buffer = new ArrayBuffer(length);
          const file = fileIo.openSync(session.cachePath, fileIo.OpenMode.READ_ONLY);
          try {
            fileIo.readSync(file.fd, buffer, { offset: start, length });
          } finally {
            fileIo.closeSync(file);
          }
          const total = session.expectedSize && session.expectedSize > 0 ? session.expectedSize : Math.max(size, fileEnd + 1);
          Logger.info(TAG, `从本地缓存返回范围 ${start}-${fileEnd}，大小: ${length} bytes`);
          return {
            buffer,
            start,
            end: fileEnd,
            total
          };
        }
      }

      const hasLocalData = size > 0;
      const requestExceedsLocal = start >= size;

      if (requestExceedsLocal) {
        if (session.protocol === 'ftp' && !hasLocalData) {
          Logger.debug(TAG, 'FTP本地无数据，直接尝试远程读取');
        }
        const remoteChunk = await this.fetchRemoteChunk(session, start, end, size);
        if (remoteChunk && remoteChunk.buffer.byteLength > 0) {
          Logger.info(TAG, `远程读取成功，返回 ${remoteChunk.buffer.byteLength} bytes，实现边下边播`);
          return remoteChunk;
        }

        if (hasLocalData) {
          Logger.info(TAG, `用户请求未下载区域 ${start}-${end}，已下载到 ${size - 1}，返回HTTP 416让播放器跳回`);
          throw new RangeNotSatisfiableError(`bytes 0-${size - 1}/${session.expectedSize || '*'}`);
        }
      }
      if (session.expectedSize && session.expectedSize > 0 && start >= session.expectedSize) {
        throw new Error('请求范围超出文件大小');
      }
      await this.delay(pollInterval);
    }
    Logger.error(TAG, `range等待超时 start=${start} end=${end} cache=${session.cachePath} protocol=${session.protocol || 'unknown'}`);
    throw new Error(`等待${session.protocol === 'ftp' ? 'FTP' : 'SMB'}缓存数据超时`);
  }

  private async tryGetFileSize(path: string): Promise<number> {
    try {
      const stat = await fileIo.stat(path);
      return stat.size;
    } catch (error) {
      return -1;
    }
  }

  private getHeader(headers: Record<string, string> | undefined, name: string): string | undefined {
    if (!headers) {
      return undefined;
    }
    const target = name.toLowerCase();
    const keys = Object.keys(headers);
    for (let i = 0; i < keys.length; i++) {
      const key = keys[i];
      if (key.toLowerCase() === target) {
        return headers[key];
      }
    }
    return undefined;
  }

  private cleanupExpiredSessions(): void {
    const now = Date.now();
    const keys = Array.from(this.sessions.keys());
    for (let i = 0; i < keys.length; i++) {
      const key = keys[i];
      const session = this.sessions.get(key);
      if (session && now - session.lastAccess > SESSION_TTL_MS) {
        this.sessions.delete(key);
        Logger.info(TAG, `移除过期SMB流会话: ${key}`);
      }
    }
  }

  private delay(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  private async fetchRemoteChunk(
    session: StreamingSession,
    start: number,
    desiredEnd: number,
    currentLocalSize: number
  ): Promise<ChunkResult | undefined> {
    if (!session.protocol || !session.remotePath) {
      return undefined;
    }
    const length = desiredEnd - start + 1;
    if (length <= 0) {
      return undefined;
    }
    try {
      let data: ArrayBuffer | undefined;
      if (session.protocol === 'smb' && session.smbConnection) {
        data = readSmbRange(session.smbConnection, session.remotePath, start, length);
      } else if (session.protocol === 'ftp' && session.ftpConnection) {
        data = await readFtpRange(session.ftpConnection, session.remotePath, start, length);
      }
      if (!data || data.byteLength === 0) {
        return undefined;
      }
      const shouldWriteToCache = currentLocalSize >= 0 && start <= currentLocalSize;
      if (shouldWriteToCache) {
        await this.writeRangeToCache(session.cachePath, start, data);
      } else {
        Logger.debug(TAG, `跳过写入缓存，避免产生空洞: start=${start}, currentLocalSize=${currentLocalSize}`);
      }
      const chunkEnd = start + data.byteLength - 1;
      const totalSize = session.expectedSize && session.expectedSize > 0 ? session.expectedSize : Math.max(chunkEnd + 1, await this.tryGetFileSize(session.cachePath));
      return {
        buffer: data,
        start,
        end: chunkEnd,
        total: totalSize
      };
    } catch (error) {
      Logger.error(TAG, `远程读取失败: ${(error as Error).message}`);
      return undefined;
    }
  }

  private async writeRangeToCache(path: string, offset: number, data: ArrayBuffer): Promise<void> {
    const file = fileIo.openSync(path, fileIo.OpenMode.CREATE | fileIo.OpenMode.READ_WRITE);
    try {
      fileIo.writeSync(file.fd, data, { offset, length: data.byteLength });
    } finally {
      fileIo.closeSync(file);
    }
  }
}