const crypto = require('crypto');
const fs = require('fs');
const path = require('path');
const { spawn } = require('child_process');

function createRtmpStreamService(options) {
  const mediaDir = options && options.mediaDir ? options.mediaDir : null;
  const ffmpegPath = options && options.ffmpegPath ? options.ffmpegPath : 'ffmpeg';
  const ffprobePath = options && options.ffprobePath ? options.ffprobePath : 'ffprobe';
  const probeIntervalMs = options && options.probeIntervalMs ? Number(options.probeIntervalMs) : 2000;
  const sessions = new Map();
  const watchedSources = new Set();
  const sourceStatusCache = new Map();
  const sourceProbePromises = new Map();
  let probeTimer = null;

  if (!mediaDir) {
    throw new Error('mediaDir is required');
  }

  const cacheRoot = path.join(mediaDir, 'rtmp-cache');

  function normalizeSourceUrl(value) {
    const sourceUrl = String(value || '').trim();
    if (!sourceUrl || !/^rtmps?:\/\//i.test(sourceUrl)) {
      const error = new Error('A valid RTMP URL is required.');
      error.statusCode = 400;
      throw error;
    }
    return sourceUrl;
  }

  function getSessionKey(sourceUrl, disableAudio) {
    return crypto.createHash('sha1').update(String(sourceUrl)).update('\0').update(disableAudio ? '1' : '0').digest('hex');
  }

  async function ensureDirectory(dirPath) {
    await fs.promises.mkdir(dirPath, { recursive: true });
  }

  async function waitForFile(filePath, timeoutMs) {
    const startedAt = Date.now();
    while (Date.now() - startedAt < timeoutMs) {
      try {
        const stat = await fs.promises.stat(filePath);
        if (stat.isFile() && stat.size > 0) {
          return true;
        }
      } catch (_error) {
        // keep waiting
      }
      await new Promise((resolve) => setTimeout(resolve, 100));
    }
    return false;
  }

  function runCommand(command, args, timeoutMs) {
    return new Promise(function (resolve) {
      var child = spawn(command, args, {
        stdio: ['ignore', 'pipe', 'pipe']
      });
      var stdout = '';
      var stderr = '';
      var finished = false;
      var timer = null;

      function done(result) {
        if (finished) {
          return;
        }
        finished = true;
        if (timer) {
          clearTimeout(timer);
          timer = null;
        }
        resolve(result);
      }

      child.stdout.on('data', function (chunk) {
        stdout += String(chunk || '');
      });

      child.stderr.on('data', function (chunk) {
        stderr += String(chunk || '');
      });

      child.on('error', function (error) {
        done({ ok: false, error: error, stdout: stdout, stderr: stderr, timedOut: false });
      });

      child.on('exit', function (code, signal) {
        done({ ok: code === 0, code: code, signal: signal, stdout: stdout, stderr: stderr, timedOut: false });
      });

      timer = setTimeout(function () {
        try {
          child.kill('SIGKILL');
        } catch (_error) {
          // ignore timeout cleanup errors
        }
        done({ ok: false, code: null, signal: 'SIGKILL', stdout: stdout, stderr: stderr, timedOut: true });
      }, Math.max(1000, Number(timeoutMs || 0) || 4000));
    });
  }

  async function probeSourceUrl(sourceUrl) {
    const normalizedSource = normalizeSourceUrl(sourceUrl);
    const result = await runCommand(ffprobePath, [
      '-hide_banner',
      '-loglevel', 'error',
      '-rw_timeout', '3000000',
      '-show_streams',
      '-of', 'json',
      normalizedSource
    ], 4000);

    if (!result.ok) {
      return {
        live: false,
        timedOut: Boolean(result.timedOut),
        stderr: String(result.stderr || '').trim()
      };
    }

    try {
      const parsed = JSON.parse(String(result.stdout || '{}'));
      const streams = Array.isArray(parsed && parsed.streams) ? parsed.streams : [];
      return {
        live: streams.length > 0,
        timedOut: false,
        stderr: String(result.stderr || '').trim()
      };
    } catch (_error) {
      return {
        live: false,
        timedOut: false,
        stderr: String(result.stderr || '').trim()
      };
    }
  }

  function getSourceStatus(sourceUrl) {
    return sourceStatusCache.get(String(sourceUrl || '').trim()) || null;
  }

  function setSourceStatus(sourceUrl, nextStatus) {
    const normalizedSource = String(sourceUrl || '').trim();
    if (!normalizedSource) {
      return null;
    }
    const status = Object.assign({
      live: false,
      probing: false,
      checkedAt: Date.now(),
      timedOut: false,
      stderr: ''
    }, nextStatus || {});
    status.live = Boolean(status.live);
    status.probing = Boolean(status.probing);
    status.checkedAt = Number(status.checkedAt || Date.now());
    sourceStatusCache.set(normalizedSource, status);
    return status;
  }

  function registerSourceWatch(sourceUrl) {
    const normalizedSource = normalizeSourceUrl(sourceUrl);
    watchedSources.add(normalizedSource);
    return normalizedSource;
  }

  async function refreshSourceStatus(sourceUrl) {
    const normalizedSource = registerSourceWatch(sourceUrl);
    const existingPromise = sourceProbePromises.get(normalizedSource) || null;
    if (existingPromise) {
      return existingPromise;
    }

    setSourceStatus(normalizedSource, Object.assign({}, getSourceStatus(normalizedSource) || {}, {
      probing: true
    }));

    const probePromise = probeSourceUrl(normalizedSource).then(function (probe) {
      return setSourceStatus(normalizedSource, {
        live: Boolean(probe.live),
        probing: false,
        checkedAt: Date.now(),
        timedOut: Boolean(probe.timedOut),
        stderr: String(probe.stderr || '')
      });
    }).catch(function (error) {
      return setSourceStatus(normalizedSource, {
        live: false,
        probing: false,
        checkedAt: Date.now(),
        timedOut: false,
        stderr: String(error && error.message ? error.message : '')
      });
    }).finally(function () {
      sourceProbePromises.delete(normalizedSource);
    });

    sourceProbePromises.set(normalizedSource, probePromise);
    return probePromise;
  }

  function scheduleSourceRefresh(sourceUrl) {
    const normalizedSource = registerSourceWatch(sourceUrl);
    const currentStatus = getSourceStatus(normalizedSource);
    if (currentStatus && currentStatus.probing) {
      return sourceProbePromises.get(normalizedSource) || Promise.resolve(currentStatus);
    }
    return refreshSourceStatus(normalizedSource);
  }

  function isSourceStatusFresh(status) {
    if (!status || !status.checkedAt) {
      return false;
    }
    return Date.now() - Number(status.checkedAt || 0) < probeIntervalMs;
  }

  function sweepWatchedSources() {
    watchedSources.forEach(function (sourceUrl) {
      const status = getSourceStatus(sourceUrl);
      if (!status || !isSourceStatusFresh(status) || status.live === false) {
        scheduleSourceRefresh(sourceUrl);
      }
    });
  }

  if (probeIntervalMs > 0) {
    probeTimer = setInterval(function () {
      sweepWatchedSources();
    }, probeIntervalMs);
    if (probeTimer && typeof probeTimer.unref === 'function') {
      probeTimer.unref();
    }
  }

  async function isSessionLive(session) {
    if (!session || !session.process) {
      return false;
    }
    if (session.exitCode !== undefined && session.exitCode !== null) {
      return false;
    }
    if (session.exitSignal !== undefined && session.exitSignal !== null) {
      return false;
    }
    try {
      const stat = await fs.promises.stat(session.manifestPath);
      return stat.isFile() && stat.size > 0;
    } catch (_error) {
      return false;
    }
  }

  function buildPlaylistUrl(key) {
    return '/api/rtmp/streams/' + encodeURIComponent(key) + '/index.m3u8';
  }

  async function ensureSession(sourceUrl, disableAudio) {
    const normalizedSource = normalizeSourceUrl(sourceUrl);
    const normalizedDisableAudio = Boolean(disableAudio);
    const key = getSessionKey(normalizedSource, normalizedDisableAudio);
    const directory = path.join(cacheRoot, key);
    const manifestPath = path.join(directory, 'index.m3u8');

    const existingSession = sessions.get(key) || null;
    if (existingSession && existingSession.process && existingSession.exitCode === null && existingSession.exitSignal === null) {
      return existingSession;
    }

    if (existingSession) {
      sessions.delete(key);
      try {
        if (existingSession.process && typeof existingSession.process.kill === 'function') {
          existingSession.process.kill('SIGKILL');
        }
      } catch (_error) {
        // ignore cleanup errors
      }
      try {
        await fs.promises.rm(directory, { recursive: true, force: true });
      } catch (_error2) {
        // ignore cleanup errors
      }
    }

    await ensureDirectory(directory);

    const args = [
      '-hide_banner',
      '-loglevel', 'warning',
      '-nostdin',
      '-i', normalizedSource,
      '-fflags', '+genpts',
      '-f', 'hls',
      '-hls_time', '1',
      '-hls_init_time', '1',
      '-hls_list_size', '8',
      '-hls_flags', 'delete_segments+append_list+omit_endlist+independent_segments+program_date_time',
      '-hls_segment_filename', path.join(directory, 'segment-%05d.ts')
    ];

    if (normalizedDisableAudio) {
      args.push('-an');
      args.push('-c:v', 'copy');
    } else {
      args.push('-c', 'copy');
    }

    args.push(manifestPath);

    const child = spawn(ffmpegPath, args, {
      stdio: ['ignore', 'ignore', 'pipe']
    });

    child.stderr.on('data', function (chunk) {
      const message = String(chunk || '').trim();
      if (message) {
        console.error('[rtmp]', message);
      }
    });

    child.on('exit', function (code, signal) {
      const session = sessions.get(key);
      if (session && session.process === child) {
        session.process = null;
        session.exitCode = code;
        session.exitSignal = signal;
      }
    });

    const session = {
      key: key,
      sourceUrl: normalizedSource,
      disableAudio: normalizedDisableAudio,
      directory: directory,
      manifestPath: manifestPath,
      playlistUrl: buildPlaylistUrl(key),
      process: child,
      exitCode: null,
      exitSignal: null,
      ready: waitForFile(manifestPath, 5000)
    };

    sessions.set(key, session);
    return session;
  }

  async function getPlaylistUrl(sourceUrl, disableAudio) {
    const session = await ensureSession(sourceUrl, disableAudio);
    await session.ready.catch(function () {
      return false;
    });
    return session.playlistUrl + '?t=' + Date.now();
  }

  async function getSessionStatus(sourceUrl, disableAudio) {
    const normalizedSource = normalizeSourceUrl(sourceUrl);
    const normalizedDisableAudio = Boolean(disableAudio);
    registerSourceWatch(normalizedSource);

    let sourceStatus = getSourceStatus(normalizedSource);
    if (!sourceStatus) {
      scheduleSourceRefresh(normalizedSource).catch(function (_error) {
        return null;
      });
      sourceStatus = getSourceStatus(normalizedSource);
    } else if (!isSourceStatusFresh(sourceStatus) && !sourceStatus.probing) {
      scheduleSourceRefresh(normalizedSource).catch(function (_error) {
        return null;
      });
    }

    if (!sourceStatus || !sourceStatus.live) {
      return {
        session: null,
        ready: false,
        live: false,
        probing: Boolean(sourceStatus && sourceStatus.probing),
        checkedAt: sourceStatus ? sourceStatus.checkedAt : null,
        timedOut: Boolean(sourceStatus && sourceStatus.timedOut),
        stderr: String(sourceStatus && sourceStatus.stderr || '')
      };
    }

    const session = await ensureSession(normalizedSource, normalizedDisableAudio);
    const ready = await session.ready.catch(function () {
      return false;
    });
    const live = ready && Boolean(session && session.process && session.exitCode === null && session.exitSignal === null);
    return {
      session: session,
      ready: ready,
      live: live,
      probing: Boolean(sourceStatus && sourceStatus.probing),
      checkedAt: sourceStatus ? sourceStatus.checkedAt : null,
      timedOut: false,
      stderr: ''
    };
  }

  function getSessionByKey(key) {
    return sessions.get(String(key || '').trim()) || null;
  }

  async function getManifestFilePath(key) {
    const session = getSessionByKey(key);
    if (!session) {
      return null;
    }
    if (!session.process || session.exitCode !== null || session.exitSignal !== null) {
      return null;
    }
    const live = await isSessionLive(session);
    if (!live) {
      return null;
    }
    return session.manifestPath;
  }

  async function getSegmentFilePath(key, fileName) {
    const session = getSessionByKey(key);
    if (!session) {
      return null;
    }
    const segmentName = path.basename(String(fileName || '').trim());
    if (!segmentName || segmentName === 'index.m3u8') {
      return null;
    }
    return path.join(session.directory, segmentName);
  }

  return {
    ensureSession: ensureSession,
    getPlaylistUrl: getPlaylistUrl,
    getSessionByKey: getSessionByKey,
    isSessionLive: isSessionLive,
    getSessionStatus: getSessionStatus,
    refreshSourceStatus: refreshSourceStatus,
    scheduleSourceRefresh: scheduleSourceRefresh,
    getSourceStatus: getSourceStatus,
    getManifestFilePath: getManifestFilePath,
    getSegmentFilePath: getSegmentFilePath
  };
}

module.exports = {
  createRtmpStreamService: createRtmpStreamService
};