#include "mpp_h264_source.h"

#include "webrtcpush_config.h"
#include "webrtcpush_log.h"

#include <gst/app/gstappsink.h>
#include <gst/video/video-event.h>
#include <fcntl.h>
#include <linux/videodev2.h>
#include <string.h>
#include <sys/ioctl.h>
#include <unistd.h>

#define MPP_VIDEO_WIDTH  1280
#define MPP_VIDEO_HEIGHT 720
#define MPP_VIDEO_GOP    WEBRTCPUSH_H264_GOP
#define MPP_VIDEO_START_BPS WEBRTCPUSH_INITIAL_BITRATE
#define MPP_VIDEO_MIN_BPS   WEBRTCPUSH_MIN_BITRATE
#define MPP_VIDEO_MAX_BPS   WEBRTCPUSH_MAX_BITRATE
#define MPP_H264_PROFILE    66  /* Baseline: 与 SDP profile-level-id=42e01f 一致，低延迟优先 */
#define MPP_IDR_EVENT_RETRY_US 500000

#define WRTC_V4L2_CAP_VIDEO_CAPTURE         (1u << 0)
#define WRTC_V4L2_CAP_VIDEO_CAPTURE_MPLANE  (1u << 12)

typedef enum {
  CAPTURE_NONE = 0,
  CAPTURE_RAW_NV12_720P,
  CAPTURE_MJPEG_720P24,
  CAPTURE_MJPEG_720P30,
} CaptureMode;

struct MppH264Source {
  GstElement *pipeline;
  GstElement *enc;
  GstElement *parse;
  GstElement *eq;
  GstElement *appsink;
  GstElement *vsrc;     /* v4l2src 指针，用于 recover 时更新设备 */
  gboolean use_mpp;
  guint8 *frame_buf;
  size_t frame_size;
  size_t frame_capacity;
  guint keyframe_req;
  volatile gboolean pending_idr;
  gboolean pending_idr_event_sent;
  gint64 last_keyframe_event_us;
  volatile gboolean pending_recover;
  gulong enc_sink_probe_id;
  guint bus_watch_id;
  gchar *sprop_parameter_sets;
};

static gboolean pull_sample(MppH264Source *src, gboolean require_idr,
                            gint timeout_ms, const uint8_t **data,
                            size_t *size, gboolean *is_idr,
                            guint64 *pts_ns);
static gboolean send_force_key_unit(MppH264Source *src);

static gboolean bus_watch_cb(GstBus *bus, GstMessage *msg, gpointer user_data)
{
  MppH264Source *src = user_data;

  (void)bus;
  if (!src)
    return TRUE;

  switch (GST_MESSAGE_TYPE(msg)) {
  case GST_MESSAGE_ERROR: {
    GError *err = NULL;
    gchar *dbg = NULL;
    gst_message_parse_error(msg, &err, &dbg);
    my_zlog_error("mpp_h264_source: pipeline error: %s (%s), recovering",
                  err ? err->message : "unknown", dbg ? dbg : "");
    g_clear_error(&err);
    g_free(dbg);
    /* ERROR 后 pipeline 进入错误态，appsink 永远拉不到帧 -> 黑屏。
     * NULL->PLAYING 重置，让 pull 线程的 recover 路径重新唤醒管道。 */
    src->pending_recover = TRUE;
    break;
  }
  case GST_MESSAGE_WARNING: {
    GError *err = NULL;
    gchar *dbg = NULL;
    gst_message_parse_warning(msg, &err, &dbg);
    my_zlog_warn("mpp_h264_source: pipeline warning: %s (%s)",
                 err ? err->message : "unknown", dbg ? dbg : "");
    g_clear_error(&err);
    g_free(dbg);
    break;
  }
  case GST_MESSAGE_EOS:
    my_zlog_warn("mpp_h264_source: pipeline EOS");
    break;
  default:
    break;
  }
  return TRUE;
}

static gchar *pick_v4l2_capture_device(void)
{
  for (int i = 0; i < 32; i++) {
    gchar *path = g_strdup_printf("/dev/video%d", i);
    if (!g_file_test(path, G_FILE_TEST_EXISTS)) {
      g_free(path);
      continue;
    }

    gchar *sys_cap = g_strdup_printf("/sys/class/video4linux/video%d/device_caps", i);
    gboolean use = FALSE;

    if (g_file_test(sys_cap, G_FILE_TEST_EXISTS)) {
      gchar *contents = NULL;
      if (g_file_get_contents(sys_cap, &contents, NULL, NULL)) {
        guint64 caps = g_ascii_strtoull(g_strstrip(contents), NULL, 0);
        use = (caps & WRTC_V4L2_CAP_VIDEO_CAPTURE) != 0 ||
              (caps & WRTC_V4L2_CAP_VIDEO_CAPTURE_MPLANE) != 0;
        g_free(contents);
      }
    } else {
      use = TRUE;
    }

    g_free(sys_cap);
    if (use)
      return path;
    g_free(path);
  }
  return NULL;
}

static guint frmival_to_fps(const struct v4l2_frmivalenum *fi)
{
  if (!fi)
    return 0;
  if (fi->type == V4L2_FRMIVAL_TYPE_DISCRETE) {
    if (!fi->discrete.denominator || !fi->discrete.numerator)
      return 0;
    return fi->discrete.denominator / fi->discrete.numerator;
  }
  if (fi->type == V4L2_FRMIVAL_TYPE_STEPWISE ||
      fi->type == V4L2_FRMIVAL_TYPE_CONTINUOUS) {
    if (!fi->stepwise.min.denominator || !fi->stepwise.min.numerator)
      return 0;
    return fi->stepwise.min.denominator / fi->stepwise.min.numerator;
  }
  return 0;
}

static gboolean frmival_meets_fps(const struct v4l2_frmivalenum *fi, guint min_fps)
{
  guint fps = frmival_to_fps(fi);
  return fps >= min_fps;
}

static gboolean size_is_720p(const struct v4l2_frmsizeenum *fs)
{
  if (fs->type == V4L2_FRMSIZE_TYPE_DISCRETE)
    return fs->discrete.width == MPP_VIDEO_WIDTH &&
           fs->discrete.height == MPP_VIDEO_HEIGHT;
  if (fs->type == V4L2_FRMSIZE_TYPE_STEPWISE ||
      fs->type == V4L2_FRMSIZE_TYPE_CONTINUOUS)
    return fs->stepwise.min_width <= MPP_VIDEO_WIDTH &&
           fs->stepwise.max_width >= MPP_VIDEO_WIDTH &&
           fs->stepwise.min_height <= MPP_VIDEO_HEIGHT &&
           fs->stepwise.max_height >= MPP_VIDEO_HEIGHT;
  return FALSE;
}

/* 探测 720p 采集：优先 NV12/YUYV@24fps+，否则 MJPEG@30（本机 USB 头常见情况） */
static CaptureMode probe_720p_capture_mode(const char *device)
{
  int fd;
  struct v4l2_fmtdesc fmt;
  gboolean raw_ok = FALSE;
  gboolean mjpg24_ok = FALSE;
  gboolean mjpg30_ok = FALSE;

  if (!device || !device[0])
    return CAPTURE_NONE;

  fd = open(device, O_RDONLY | O_NONBLOCK);
  if (fd < 0)
    return CAPTURE_NONE;

  memset(&fmt, 0, sizeof(fmt));
  fmt.type = V4L2_BUF_TYPE_VIDEO_CAPTURE;

  for (; ioctl(fd, VIDIOC_ENUM_FMT, &fmt) == 0; fmt.index++) {
    struct v4l2_frmsizeenum fs;

    memset(&fs, 0, sizeof(fs));
    fs.pixel_format = fmt.pixelformat;
    fs.index = 0;

    for (; ioctl(fd, VIDIOC_ENUM_FRAMESIZES, &fs) == 0; fs.index++) {
      struct v4l2_frmivalenum fi;

      if (!size_is_720p(&fs))
        continue;

      memset(&fi, 0, sizeof(fi));
      fi.pixel_format = fmt.pixelformat;
      fi.index = 0;
      if (fs.type == V4L2_FRMSIZE_TYPE_DISCRETE) {
        fi.width = fs.discrete.width;
        fi.height = fs.discrete.height;
      } else {
        fi.width = MPP_VIDEO_WIDTH;
        fi.height = MPP_VIDEO_HEIGHT;
      }

      for (; ioctl(fd, VIDIOC_ENUM_FRAMEINTERVALS, &fi) == 0; fi.index++) {
        guint fps = frmival_to_fps(&fi);

        if (fmt.pixelformat == V4L2_PIX_FMT_MJPEG) {
          /* 必须能直出目标帧率；30fps 不能冒充 24fps（v4l2 会 not-negotiated） */
          if (fps == WEBRTCPUSH_H264_FPS)
            mjpg24_ok = TRUE;
          if (fps >= 30)
            mjpg30_ok = TRUE;
        }
        if ((fmt.pixelformat == V4L2_PIX_FMT_NV12 ||
             fmt.pixelformat == V4L2_PIX_FMT_YUYV) &&
            fps >= WEBRTCPUSH_H264_FPS) {
          raw_ok = TRUE;
        }
      }
    }
  }

  close(fd);

  if (raw_ok) {
    my_zlog_info("mpp_h264_source: %s supports 720p raw >=%dfps, use NV12/YUYV direct",
                 device, WEBRTCPUSH_H264_FPS);
    return CAPTURE_RAW_NV12_720P;
  }
  if (mjpg24_ok) {
    my_zlog_info("mpp_h264_source: %s supports 720p MJPEG@%dfps direct",
                 device, WEBRTCPUSH_H264_FPS);
    return CAPTURE_MJPEG_720P24;
  }
  if (mjpg30_ok) {
    my_zlog_info("mpp_h264_source: %s 720p MJPEG@30fps -> mppjpegdec -> videorate@%dfps",
                 device, WEBRTCPUSH_H264_FPS);
    return CAPTURE_MJPEG_720P30;
  }

  my_zlog_warn("mpp_h264_source: %s has no suitable 720p capture mode", device);
  return CAPTURE_NONE;
}

static void configure_pre_enc_queue(GstElement *q)
{
  if (!q)
    return;
  g_object_set(q,
      "max-size-buffers", WEBRTCPUSH_MPP_PRE_ENC_BUFFERS,
      "max-size-bytes", 0,
      "max-size-time", (guint64)GST_SECOND * 2 / WEBRTCPUSH_H264_FPS,
      "leaky", 2,
      NULL);
}

static void configure_post_enc_queue(GstElement *q)
{
  if (!q)
    return;
  g_object_set(q,
      "max-size-buffers", WEBRTCPUSH_MPP_POST_ENC_BUFFERS,
      "max-size-bytes", 0,
      "max-size-time", (guint64)GST_SECOND / WEBRTCPUSH_H264_FPS,
      "leaky", 0,
      NULL);
}

static void configure_v4l2_src(GstElement *src, gboolean compressed_mjpeg)
{
  if (!src)
    return;
  g_object_set(src, "do-timestamp", TRUE, NULL);
  /*
   * MJPEG 压缩流用 mmap；原始 NV12/YUYV 统一用 mmap（io-mode=2）。
   * DMABUF(4) 在非 RK 平台/无 dma-buf 导出的 USB 摄像头上会 negotiation 失败 -> 黑屏。
   * mmap 兼容性最好，性能损失可接受（720p@24fps）。
   */
  g_object_set(src, "io-mode", 2, NULL);
  /* 减少内核 DQBUF 缓冲，降低采集延迟（部分 v4l2src 无此属性，先检查） */
  {
    GParamSpec *pspec = g_object_class_find_property(G_OBJECT_GET_CLASS(src), "buffer-size");
    if (pspec)
      g_object_set(src, "buffer-size", 1024, NULL);
  }
}

static GstElement *make_h264_encoder(gboolean prefer_mpp, gboolean use_test,
                                     gboolean *use_mpp_out)
{
  GstElement *enc = NULL;

  if (prefer_mpp && !use_test) {
    enc = gst_element_factory_make("mpph264enc", "enc");
    if (enc) {
      *use_mpp_out = TRUE;
      g_object_set(enc,
          "gop", MPP_VIDEO_GOP,
          "rc-mode", 1,  /* CBR */
          "bps", MPP_VIDEO_START_BPS,
          "bps-min", (guint)(((guint64)MPP_VIDEO_START_BPS *
                              WEBRTCPUSH_MPP_BPS_MIN_PERCENT) / 100U),
          "bps-max", (guint)(((guint64)MPP_VIDEO_START_BPS *
                              WEBRTCPUSH_MPP_BPS_MAX_PERCENT) / 100U),
          "header-mode", 1,
          "profile", MPP_H264_PROFILE,
          "level", 31,
          "qp-min", WEBRTCPUSH_MPP_QP_MIN,
          "qp-min-i", WEBRTCPUSH_MPP_QP_MIN_I,
          "qp-max", WEBRTCPUSH_MPP_QP_MAX,
          "qp-max-i", WEBRTCPUSH_MPP_QP_MAX_I,
          "min-force-key-unit-interval",
              (guint64)WEBRTCPUSH_MPP_MIN_FORCE_KEY_UNIT_NS,
          "qos", TRUE,
          "zero-copy-pkt", TRUE,
          NULL);
      my_zlog_info("mpp_h264_source: mpph264enc %dx%d@%dfps GOP=%d Baseline/CBR",
                   MPP_VIDEO_WIDTH, MPP_VIDEO_HEIGHT, WEBRTCPUSH_H264_FPS,
                   MPP_VIDEO_GOP);
      return enc;
    }
    my_zlog_warn("mpp_h264_source: mpph264enc unavailable");
  }

  enc = gst_element_factory_make("x264enc", "enc");
  if (enc) {
    g_object_set(enc,
        "tune", "zerolatency",
        "speed-preset", "ultrafast",
        "bitrate", MPP_VIDEO_START_BPS / 1000,
        "key-int-max", MPP_VIDEO_GOP,
        "byte-stream", TRUE,
        "profile", "baseline",
        NULL);
    my_zlog_info("mpp_h264_source: x264enc software fallback (baseline, matches SDP)");
    return enc;
  }

  return NULL;
}

static gboolean buffer_is_idr(const guint8 *data, size_t size)
{
  size_t i = 0;

  while (i + 3 < size) {
    size_t sc = 0;
    guint8 nal_type;

    if (i + 4 <= size && data[i] == 0 && data[i + 1] == 0 &&
        data[i + 2] == 0 && data[i + 3] == 1)
      sc = 4;
    else if (data[i] == 0 && data[i + 1] == 0 && data[i + 2] == 1)
      sc = 3;
    else {
      i++;
      continue;
    }

    i += sc;
    if (i >= size)
      break;
    nal_type = data[i] & 0x1f;
    if (nal_type == 5)
      return TRUE;
    if (nal_type == 1)
      return FALSE;
  }
  return FALSE;
}

static GstPadProbeReturn enc_sink_idr_probe(GstPad *pad, GstPadProbeInfo *info,
                                            gpointer user_data)
{
  MppH264Source *src = user_data;

  (void)pad;
  if (!(GST_PAD_PROBE_INFO_TYPE(info) & GST_PAD_PROBE_TYPE_BUFFER))
    return GST_PAD_PROBE_OK;
  if (!src || !src->pending_idr)
    return GST_PAD_PROBE_OK;

  {
    gint64 now_us = g_get_monotonic_time();
    if (src->pending_idr_event_sent &&
        now_us - src->last_keyframe_event_us < MPP_IDR_EVENT_RETRY_US)
      return GST_PAD_PROBE_OK;

    src->last_keyframe_event_us = now_us;
    if (send_force_key_unit(src)) {
      src->pending_idr_event_sent = TRUE;
      my_zlog_info("mpp_h264_source: IDR request sent (%u)", src->keyframe_req);
    } else {
      src->pending_idr_event_sent = FALSE;
      my_zlog_warn("mpp_h264_source: in-stream IDR request not handled");
    }
  }
  return GST_PAD_PROBE_OK;
}

static gboolean send_force_key_unit(MppH264Source *src)
{
  GstPad *src_pad;
  GstEvent *ev;
  GstClockTime running_time = 0;
  gboolean ok = FALSE;

  if (!src || !src->enc)
    return FALSE;

  if (src->pipeline) {
    GstClock *clock = gst_element_get_clock(src->pipeline);
    if (clock) {
      GstClockTime now = gst_clock_get_time(clock);
      GstClockTime base = gst_element_get_base_time(src->pipeline);
      if (GST_CLOCK_TIME_IS_VALID(now) && GST_CLOCK_TIME_IS_VALID(base) &&
          now >= base)
        running_time = now - base;
      gst_object_unref(clock);
    }
  }

  /*
   * force-key-unit 是 upstream event，应从编码器 src pad 往上游发送。
   * 发到 sink pad 会触发 wrong-direction，并且 PLI 后经常等不到真正 IDR。
   */
  src_pad = gst_element_get_static_pad(src->enc, "src");
  if (src_pad) {
    ev = gst_video_event_new_upstream_force_key_unit(
        running_time, TRUE, src->keyframe_req);
    ok = gst_pad_send_event(src_pad, ev);
    gst_object_unref(src_pad);
  }

  if (!ok) {
    ev = gst_video_event_new_upstream_force_key_unit(
        running_time, TRUE, src->keyframe_req);
    ok = gst_element_send_event(src->enc, ev);
  }

  return ok;
}

static void request_keyframe(MppH264Source *src)
{
  if (!src)
    return;
  if (src->pending_idr)
    return;

  src->keyframe_req++;
  src->pending_idr = TRUE;
  src->pending_idr_event_sent = FALSE;
  src->last_keyframe_event_us = 0;
  /* IDR 由 enc_sink_idr_probe 在下一个 buffer 经过时发送，
     避免外部直接发 sink pad 触发 wrong-direction 警告 */
  my_zlog_info("mpp_h264_source: IDR queued (%u)", src->keyframe_req);
}

void mpp_h264_source_request_keyframe(MppH264Source *src)
{
  request_keyframe(src);
}

static gboolean extract_sprop_from_au(MppH264Source *src, const guint8 *data,
                                      size_t size)
{
  gchar *sps_b64 = NULL;
  gchar *pps_b64 = NULL;

  if (!src || !data || size < 5 || src->sprop_parameter_sets)
    return src && src->sprop_parameter_sets != NULL;

  for (size_t i = 0; i + 4 < size;) {
    size_t nal_start = 0;
    guint8 nal_type;

    if (data[i] == 0 && data[i + 1] == 0 && data[i + 2] == 1) {
      nal_start = i + 3;
      i += 3;
    } else if (data[i] == 0 && data[i + 1] == 0 && data[i + 2] == 0 &&
               data[i + 3] == 1) {
      nal_start = i + 4;
      i += 4;
    } else {
      i++;
      continue;
    }

    nal_type = data[nal_start] & 0x1f;
    {
      size_t j = nal_start + 1;
      while (j + 3 < size) {
        if (data[j] == 0 && data[j + 1] == 0 &&
            (data[j + 2] == 1 || (data[j + 2] == 0 && data[j + 3] == 1)))
          break;
        j++;
      }
      if (j + 3 >= size)
        j = size;
      if (nal_type == 7 && !sps_b64)
        sps_b64 = g_base64_encode(data + nal_start, j - nal_start);
      else if (nal_type == 8 && !pps_b64)
        pps_b64 = g_base64_encode(data + nal_start, j - nal_start);
      i = j;
    }
    if (sps_b64 && pps_b64)
      break;
  }

  if (sps_b64 && pps_b64) {
    g_free(src->sprop_parameter_sets);
    src->sprop_parameter_sets = g_strdup_printf("%s,%s", sps_b64, pps_b64);
    my_zlog_info("mpp_h264_source: captured sprop for SDP (%zu+%zu b64)",
                 strlen(sps_b64), strlen(pps_b64));
  }
  g_free(sps_b64);
  g_free(pps_b64);
  return src->sprop_parameter_sets != NULL;
}

static gboolean copy_frame(MppH264Source *src, GstBuffer *buffer)
{
  GstMapInfo map;

  if (!gst_buffer_map(buffer, &map, GST_MAP_READ))
    return FALSE;

  if (map.size > src->frame_capacity) {
    src->frame_buf = g_realloc(src->frame_buf, map.size);
    src->frame_capacity = map.size;
  }
  memcpy(src->frame_buf, map.data, map.size);
  src->frame_size = map.size;
  extract_sprop_from_au(src, src->frame_buf, src->frame_size);
  gst_buffer_unmap(buffer, &map);
  return TRUE;
}

static gboolean link_pipeline(GstElement *pipe, GstElement **elements, guint count)
{
  guint i;

  if (!pipe || !elements || count < 2)
    return FALSE;

  for (i = 0; i < count; i++)
    gst_bin_add(GST_BIN(pipe), elements[i]);

  for (i = 0; i + 1 < count; i++) {
    if (!gst_element_link(elements[i], elements[i + 1]))
      return FALSE;
  }
  return TRUE;
}

MppH264Source *mpp_h264_source_start(const char *video_device, char **error_message)
{
  MppH264Source *src;
  GstElement *pipe, *vsrc, *caps_in = NULL, *jpegparse = NULL, *jpegdec = NULL;
  GstElement *conv = NULL, *vrate, *caps_nv12, *enc, *parse, *q, *eq, *caps_h264, *sink;
  gchar *auto_v4l = NULL;
  const gchar *vdev = video_device;
  gboolean use_test = FALSE;
  gboolean use_auto = FALSE;
  gboolean use_mpp = FALSE;
  gboolean use_mpp_jpeg = FALSE;
  gint mjpeg_fps = WEBRTCPUSH_H264_FPS;
  CaptureMode capture_mode = CAPTURE_NONE;
  GstStateChangeReturn ret;

  if (!gst_is_initialized())
    gst_init(NULL, NULL);

  src = g_new0(MppH264Source, 1);

  use_test = (vdev && g_strcmp0(vdev, "test") == 0);
  use_auto = (!vdev || g_strcmp0(vdev, "auto") == 0);

  if (use_auto && !use_test) {
    auto_v4l = pick_v4l2_capture_device();
    if (auto_v4l) {
      my_zlog_info("mpp_h264_source: auto V4L2 %s", auto_v4l);
      vdev = auto_v4l;
      capture_mode = probe_720p_capture_mode(vdev);
    } else {
      my_zlog_warn("mpp_h264_source: no V4L2 device, videotestsrc");
      use_test = TRUE;
    }
  } else if (!use_test) {
    capture_mode = probe_720p_capture_mode(vdev);
  }

  if (!use_test && capture_mode == CAPTURE_NONE) {
    if (error_message)
      *error_message = g_strdup("no 720p capture mode on camera");
    g_free(auto_v4l);
    g_free(src);
    return NULL;
  }

  pipe = gst_pipeline_new("mpp-h264-pipe");
  vrate = gst_element_factory_make("videorate", "vrate");
  caps_nv12 = gst_element_factory_make("capsfilter", "caps_nv12");
  q = gst_element_factory_make("queue", "vq");
  eq = gst_element_factory_make("queue", "eq");
  enc = make_h264_encoder(TRUE, use_test, &use_mpp);
  parse = gst_element_factory_make("h264parse", "parse");
  caps_h264 = gst_element_factory_make("capsfilter", "caps_h264");
  sink = gst_element_factory_make("appsink", "sink");

  if (use_test) {
    vsrc = gst_element_factory_make("videotestsrc", "vsrc");
    g_object_set(vsrc, "is-live", TRUE, NULL);
    conv = gst_element_factory_make("videoconvert", "conv");
  } else if (capture_mode == CAPTURE_RAW_NV12_720P) {
    vsrc = gst_element_factory_make("v4l2src", "vsrc");
    caps_in = gst_element_factory_make("capsfilter", "caps_in");
    g_object_set(vsrc, "device", vdev, NULL);
    configure_v4l2_src(vsrc, FALSE);
    {
      gchar *cap_str = g_strdup_printf(
          "video/x-raw,format=NV12,width=%d,height=%d,framerate=%d/1",
          MPP_VIDEO_WIDTH, MPP_VIDEO_HEIGHT, WEBRTCPUSH_H264_FPS);
      GstCaps *caps = gst_caps_from_string(cap_str);
      g_object_set(caps_in, "caps", caps, NULL);
      gst_caps_unref(caps);
      g_free(cap_str);
    }
    my_zlog_info("mpp_h264_source: 720p NV12 direct capture (no JPEG decode)");
  } else {
    vsrc = gst_element_factory_make("v4l2src", "vsrc");
    caps_in = gst_element_factory_make("capsfilter", "caps_in");
    jpegparse = gst_element_factory_make("jpegparse", "jpegparse");
    if (WEBRTCPUSH_MJPEG_CPU_DECODE) {
      jpegdec = gst_element_factory_make("jpegdec", "jpegdec");
      use_mpp_jpeg = FALSE;
      if (jpegdec) {
        conv = gst_element_factory_make("videoconvert", "jpegconv");
        g_object_set(jpegdec,
            "idct-method", 1,              /* ifast */
            "discard-corrupted-frames", TRUE,
            "qos", TRUE,
            NULL);
      } else {
        jpegdec = gst_element_factory_make("mppjpegdec", "jpegdec");
        if (jpegdec) {
          use_mpp_jpeg = TRUE;
          g_object_set(jpegdec, "format", 23, "fast-mode", TRUE, NULL);
        }
      }
    } else {
      jpegdec = gst_element_factory_make("mppjpegdec", "jpegdec");
      if (jpegdec) {
        use_mpp_jpeg = TRUE;
        g_object_set(jpegdec, "format", 23, "fast-mode", TRUE, NULL);
      } else {
        jpegdec = gst_element_factory_make("jpegdec", "jpegdec");
        use_mpp_jpeg = FALSE;
        if (jpegdec) {
          conv = gst_element_factory_make("videoconvert", "jpegconv");
          g_object_set(jpegdec,
              "idct-method", 1,
              "discard-corrupted-frames", TRUE,
              "qos", TRUE,
              NULL);
        }
      }
    }
    mjpeg_fps = (capture_mode == CAPTURE_MJPEG_720P24) ? WEBRTCPUSH_H264_FPS : 30;
    g_object_set(vsrc, "device", vdev, NULL);
    configure_v4l2_src(vsrc, TRUE);
    {
      gchar *cap_str = g_strdup_printf(
          "image/jpeg,width=%d,height=%d,framerate=%d/1",
          MPP_VIDEO_WIDTH, MPP_VIDEO_HEIGHT, mjpeg_fps);
      GstCaps *caps = gst_caps_from_string(cap_str);
      g_object_set(caps_in, "caps", caps, NULL);
      gst_caps_unref(caps);
      g_free(cap_str);
    }
    my_zlog_info("mpp_h264_source: 720p MJPEG@%dfps -> %s%s -> NV12@%dfps",
                 mjpeg_fps,
                 use_mpp_jpeg ? "mppjpegdec" : "jpegdec",
                 use_mpp_jpeg ? "" : "+videoconvert",
                 WEBRTCPUSH_H264_FPS);
  }

  if (!pipe || !vsrc || !vrate || !caps_nv12 || !q || !eq || !enc || !parse ||
      !caps_h264 || !sink) {
    if (error_message)
      *error_message = g_strdup("GStreamer element create failed");
    g_free(auto_v4l);
    g_free(src);
    return NULL;
  }

  g_object_set(vrate, "max-rate", WEBRTCPUSH_H264_FPS, "drop-only", TRUE, NULL);
  configure_pre_enc_queue(q);
  configure_post_enc_queue(eq);
  g_object_set(parse, "config-interval", -1, NULL);

  {
    gchar *cap_str = g_strdup_printf(
        "video/x-raw,format=NV12,width=%d,height=%d,framerate=%d/1",
        use_test ? 640 : MPP_VIDEO_WIDTH,
        use_test ? 480 : MPP_VIDEO_HEIGHT,
        WEBRTCPUSH_H264_FPS);
    GstCaps *caps = gst_caps_from_string(cap_str);
    g_object_set(caps_nv12, "caps", caps, NULL);
    gst_caps_unref(caps);
    g_free(cap_str);
  }

  {
    GstCaps *hc = gst_caps_from_string(
        "video/x-h264,stream-format=byte-stream,alignment=au");
    g_object_set(caps_h264, "caps", hc, NULL);
    gst_caps_unref(hc);
  }

  g_object_set(sink,
      "emit-signals", FALSE,
      "sync", FALSE,
      "max-buffers", WEBRTCPUSH_APPSINK_MAX_BUFFERS,
      "drop", TRUE,
      NULL);

  if (use_test) {
    GstElement *elems[] = {vsrc, conv, vrate, caps_nv12, q, enc, parse, caps_h264, eq, sink};
    if (!link_pipeline(pipe, elems, G_N_ELEMENTS(elems))) {
      if (error_message)
        *error_message = g_strdup("failed to link test chain");
      gst_object_unref(pipe);
      g_free(auto_v4l);
      g_free(src);
      return NULL;
    }
  } else if (capture_mode == CAPTURE_RAW_NV12_720P) {
    GstElement *elems[] = {vsrc, caps_in, vrate, caps_nv12, q, enc, parse, caps_h264, eq, sink};
    if (!link_pipeline(pipe, elems, G_N_ELEMENTS(elems))) {
      if (error_message)
        *error_message = g_strdup("failed to link raw NV12 chain");
      gst_object_unref(pipe);
      g_free(auto_v4l);
      g_free(src);
      return NULL;
    }
  } else {
    gboolean linked;
    if (use_mpp_jpeg) {
      GstElement *elems[] = {vsrc, caps_in, jpegparse, jpegdec, vrate,
                             caps_nv12, q, enc, parse, caps_h264, eq, sink};
      linked = jpegparse && jpegdec &&
          link_pipeline(pipe, elems, G_N_ELEMENTS(elems));
    } else {
      GstElement *elems[] = {vsrc, caps_in, jpegparse, jpegdec, conv, vrate,
                             caps_nv12, q, enc, parse, caps_h264, eq, sink};
      linked = jpegparse && jpegdec && conv &&
          link_pipeline(pipe, elems, G_N_ELEMENTS(elems));
    }
    if (!linked) {
      if (error_message)
        *error_message = g_strdup("failed to link MJPEG 720p chain");
      gst_object_unref(pipe);
      g_free(auto_v4l);
      g_free(src);
      return NULL;
    }
  }

  src->vsrc = vsrc;
  src->pipeline = pipe;
  src->enc = enc;
  src->parse = parse;
  src->eq = eq;
  src->appsink = sink;
  src->use_mpp = use_mpp;

  {
    GstPad *enc_sink = gst_element_get_static_pad(enc, "sink");
    if (enc_sink) {
      src->enc_sink_probe_id = gst_pad_add_probe(
          enc_sink, GST_PAD_PROBE_TYPE_BUFFER, enc_sink_idr_probe, src, NULL);
      gst_object_unref(enc_sink);
    }
  }

  gst_pipeline_set_latency(GST_PIPELINE(pipe), 0);
  ret = gst_element_set_state(pipe, GST_STATE_PLAYING);
  if (ret == GST_STATE_CHANGE_FAILURE) {
    if (error_message)
      *error_message = g_strdup("pipeline failed to reach PLAYING");
    mpp_h264_source_stop(src);
    g_free(auto_v4l);
    return NULL;
  }

  {
    GstBus *bus = gst_element_get_bus(pipe);
    if (bus) {
      src->bus_watch_id = gst_bus_add_watch(bus, bus_watch_cb, src);
      gst_object_unref(bus);
    }
  }

  {
    const guint8 *warm_data;
    size_t warm_size;
    gboolean warm_idr;

    if (!pull_sample(src, FALSE, 8000, &warm_data, &warm_size, &warm_idr, NULL)) {
      if (error_message)
        *error_message = g_strdup("encoder warmup timeout (no frame in 8s)");
      my_zlog_error("mpp_h264_source: warmup failed, check camera /dev/video*");
      mpp_h264_source_stop(src);
      g_free(auto_v4l);
      return NULL;
    }
    my_zlog_info("mpp_h264_source: warmup OK first_au=%zu bytes idr=%d",
                 warm_size, warm_idr ? 1 : 0);
  }

  my_zlog_info("mpp_h264_source: live encoder started 720p (mpp=%d pre_buf=%d post_buf=%d)",
               use_mpp ? 1 : 0, WEBRTCPUSH_MPP_PRE_ENC_BUFFERS,
               WEBRTCPUSH_MPP_POST_ENC_BUFFERS);
  g_free(auto_v4l);
  return src;
}

static gboolean pull_sample(MppH264Source *src, gboolean require_idr,
                            gint timeout_ms, const uint8_t **data,
                            size_t *size, gboolean *is_idr,
                            guint64 *pts_ns)
{
  GstAppSink *appsink;
  GstSample *sample;
  GstBuffer *buffer;
  gboolean idr;
  GstClockTime pts;
  gint64 deadline;
  gint attempts = 0;

  if (!src || !src->appsink || !data || !size || timeout_ms < 1)
    return FALSE;

  appsink = GST_APP_SINK(src->appsink);
  deadline = g_get_monotonic_time() + (gint64)timeout_ms * 1000;

  while (attempts < 120) {  /* require_idr 模式下用 deadline 控制超时，不靠 attempts */
    gint wait_ms = (gint)((deadline - g_get_monotonic_time()) / 1000);

    if (wait_ms < 1)
      break;

    sample = gst_app_sink_try_pull_sample(appsink, (guint64)wait_ms * GST_MSECOND);
    if (!sample) {
      if (g_get_monotonic_time() >= deadline)
        break;
      continue;
    }

    buffer = gst_sample_get_buffer(sample);
    if (!buffer || !copy_frame(src, buffer)) {
      gst_sample_unref(sample);
      attempts++;
      continue;
    }

    pts = GST_BUFFER_PTS(buffer);
    idr = buffer_is_idr(src->frame_buf, src->frame_size);
    if (idr && src->pending_idr) {
      src->pending_idr = FALSE;
      src->pending_idr_event_sent = FALSE;
      src->last_keyframe_event_us = 0;
      my_zlog_info("mpp_h264_source: IDR output confirmed (%u)", src->keyframe_req);
    }
    gst_sample_unref(sample);

    if (require_idr && !idr) {
      attempts++;
      continue;
    }

    *data = src->frame_buf;
    *size = src->frame_size;
    if (is_idr)
      *is_idr = idr;
    if (pts_ns)
      *pts_ns = GST_CLOCK_TIME_IS_VALID(pts) ? (guint64)pts : G_MAXUINT64;
    return TRUE;
  }

  return FALSE;
}

gboolean mpp_h264_source_pull(MppH264Source *src, gboolean force_idr,
                              const uint8_t **data, size_t *size,
                              gboolean *is_idr, guint64 *pts_ns,
                              gint timeout_ms)
{
  gint idr_timeout;

  if (!src || !src->appsink || !data || !size)
    return FALSE;

  /* bus 报 ERROR 后 pipeline 卡死，这里及时恢复避免长期黑屏 */
  if (src->pending_recover && mpp_h264_source_recover(src)) {
    src->pending_idr = TRUE;
    src->pending_idr_event_sent = FALSE;
    src->last_keyframe_event_us = 0;
  }

  if (timeout_ms < 1)
    timeout_ms = (1000 / WEBRTCPUSH_H264_FPS) + 60;

  if (force_idr) {
    idr_timeout = timeout_ms < 2000 ? 2000 : timeout_ms;
    request_keyframe(src);
    if (pull_sample(src, TRUE, idr_timeout, data, size, is_idr, pts_ns))
      return TRUE;
    my_zlog_warn("mpp_h264_source: IDR wait timeout, drain latest frame");
    /* 超时后清 pending_idr，避免无限重试导致卡死；下一个 PLI 会重新请求 */
    src->pending_idr = FALSE;
    src->pending_idr_event_sent = FALSE;
    src->last_keyframe_event_us = 0;
    if (pull_sample(src, FALSE, 500, data, size, is_idr, pts_ns))
      return TRUE;
    return FALSE;
  }

  return pull_sample(src, FALSE, timeout_ms, data, size, is_idr, pts_ns);
}

void mpp_h264_source_set_bitrate(MppH264Source *src, guint bitrate_bps)
{
  GstElementFactory *f;
  const gchar *name;

  if (!src || !src->enc || !bitrate_bps)
    return;

  if (bitrate_bps < MPP_VIDEO_MIN_BPS)
    bitrate_bps = MPP_VIDEO_MIN_BPS;
  if (bitrate_bps > MPP_VIDEO_MAX_BPS)
    bitrate_bps = MPP_VIDEO_MAX_BPS;

  f = gst_element_get_factory(src->enc);
  name = f ? gst_plugin_feature_get_name(GST_PLUGIN_FEATURE(f)) : NULL;
  if (name && g_str_has_prefix(name, "mpph264enc")) {
    guint bmin = (guint)(((guint64)bitrate_bps *
                          WEBRTCPUSH_MPP_BPS_MIN_PERCENT) / 100U);
    guint bmax = (guint)(((guint64)bitrate_bps *
                          WEBRTCPUSH_MPP_BPS_MAX_PERCENT) / 100U);
    if (bmin < WEBRTCPUSH_MPP_BPS_MIN_FLOOR)
      bmin = WEBRTCPUSH_MPP_BPS_MIN_FLOOR;
    if (bmax < bitrate_bps)
      bmax = bitrate_bps;
    if (bmax > MPP_VIDEO_MAX_BPS)
      bmax = MPP_VIDEO_MAX_BPS;
    g_object_set(src->enc,
        "bps", bitrate_bps,
        "bps-min", bmin,
        "bps-max", bmax,
        NULL);
  } else if (name && g_str_has_prefix(name, "x264enc")) {
    g_object_set(src->enc, "bitrate", bitrate_bps / 1000, NULL);
  }
}

gboolean mpp_h264_source_recover(MppH264Source *src)
{
  const guint8 *data;
  size_t size;
  gboolean idr;
  GstStateChangeReturn ret;

  if (!src || !src->pipeline)
    return FALSE;

  my_zlog_warn("mpp_h264_source: recovering pipeline (NULL -> PLAYING)");
  src->pending_idr = FALSE;
  src->pending_recover = FALSE;
  gst_element_set_state(src->pipeline, GST_STATE_NULL);
  /* 重新探测摄像头设备（USB 重枚举后可能从 video0 变成 video1/2） */
  if (src->vsrc) {
    gchar *new_dev = pick_v4l2_capture_device();
    if (new_dev) {
      gchar *old_dev = NULL;
      g_object_get(src->vsrc, "device", &old_dev, NULL);
      if (!old_dev || g_strcmp0(old_dev, new_dev) != 0) {
        g_object_set(src->vsrc, "device", new_dev, NULL);
        my_zlog_warn("mpp_h264_source: recover device changed %s -> %s",
                     old_dev ? old_dev : "(null)", new_dev);
      }
      g_free(old_dev);
      g_free(new_dev);
    }
  }
  g_usleep(300000);
  ret = gst_element_set_state(src->pipeline, GST_STATE_PLAYING);
  if (ret == GST_STATE_CHANGE_FAILURE) {
    my_zlog_error("mpp_h264_source: recover failed to reach PLAYING");
    return FALSE;
  }
  if (!pull_sample(src, FALSE, 5000, &data, &size, &idr, NULL)) {
    my_zlog_error("mpp_h264_source: recover warmup failed");
    return FALSE;
  }
  my_zlog_info("mpp_h264_source: recover OK first_au=%zu bytes", size);
  return TRUE;
}

gboolean mpp_h264_source_is_mpp_encoder(const MppH264Source *src)
{
  return src && src->use_mpp;
}

const gchar *mpp_h264_source_get_sprop(const MppH264Source *src)
{
  return src ? src->sprop_parameter_sets : NULL;
}

void mpp_h264_source_stop(MppH264Source *src)
{
  if (!src)
    return;

  if (src->bus_watch_id) {
    g_source_remove(src->bus_watch_id);
    src->bus_watch_id = 0;
  }
  if (src->pipeline) {
    gst_element_set_state(src->pipeline, GST_STATE_NULL);
    gst_object_unref(src->pipeline);
    src->pipeline = NULL;
  }
  src->enc = NULL;
  src->parse = NULL;
  src->eq = NULL;
  src->appsink = NULL;
  g_free(src->sprop_parameter_sets);
  g_free(src->frame_buf);
  g_free(src);
}
