#include "mpp_h264_source.h"

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

#include <gst/app/gstappsink.h>
#include <gst/app/gstappsrc.h>
#include <gst/video/video-event.h>
#include <fcntl.h>
#include <stdio.h>
#include <jpeglib.h>
#include <linux/videodev2.h>
#include <setjmp.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 MJPEG_DECODE_WARN_US 15000
#define MJPEG_DECODE_WARN_INTERVAL_US 5000000
#define PERF_LOG_FRAMES (WEBRTCPUSH_H264_FPS * 5)
#define TIMING_RING_SIZE 32

#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;

typedef struct {
  guint64 pts_ns;
  gint64 capture_us;
  gint64 push_us;
  gint64 enc_in_us;
  guint decode_us;
  guint convert_us;
  gboolean valid;
} FrameTiming;

struct MppH264Source {
  GstElement *pipeline;
  GstElement *capture_pipeline;
  GstElement *enc;
  GstElement *parse;
  GstElement *eq;
  GstElement *appsink;
  GstElement *mjpeg_sink;
  GstElement *raw_appsrc;
  GstElement *vsrc;     /* v4l2src 指针，用于 recover 时更新设备 */
  gboolean use_mpp;
  gboolean manual_mjpeg;
  volatile gboolean stop_decode;
  GThread *decode_thread;
  guint8 *frame_buf;
  size_t frame_size;
  size_t frame_capacity;
  guint8 *i420_buf;
  size_t i420_capacity;
  guint8 *nv12_buf;
  size_t nv12_capacity;
  guint8 *jpeg_tmp_u;
  guint8 *jpeg_tmp_v;
  size_t jpeg_tmp_capacity;
  guint keyframe_req;
  gint pending_bitrate_bps;
  guint applied_bitrate_bps;
  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;
  guint capture_bus_watch_id;
  gchar *sprop_parameter_sets;
  GMutex timing_lock;
  FrameTiming timings[TIMING_RING_SIZE];
  guint timing_next;
  guint perf_frames;
  guint64 perf_decode_sum_us;
  guint64 perf_convert_sum_us;
  guint64 perf_encode_sum_us;
  guint64 perf_total_sum_us;
  guint perf_decode_max_us;
  guint perf_convert_max_us;
  guint perf_encode_max_us;
  guint perf_total_max_us;
  guint jpeg_slow_count;
  guint jpeg_slow_last_us;
  guint jpeg_slow_max_us;
  gint64 jpeg_slow_last_warn_us;
};

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 void apply_pending_bitrate(MppH264Source *src);
static gpointer mjpeg_decode_thread(gpointer user_data);
static void timing_mark_encoder_in(MppH264Source *src, guint64 pts_ns,
                                   gint64 enc_in_us);

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 -> libjpeg-turbo -> 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 object_set_uint_if_exists(GObject *obj, const gchar *name, guint value)
{
  if (obj && g_object_class_find_property(G_OBJECT_GET_CLASS(obj), name))
    g_object_set(obj, name, value, NULL);
}

static void object_set_int_if_exists(GObject *obj, const gchar *name, gint value)
{
  if (obj && g_object_class_find_property(G_OBJECT_GET_CLASS(obj), name))
    g_object_set(obj, name, value, NULL);
}

static gboolean ensure_byte_buffer(guint8 **buf, size_t *capacity, size_t need)
{
  if (!buf || !capacity || need == 0)
    return FALSE;
  if (*capacity >= need && *buf)
    return TRUE;
  *buf = g_realloc(*buf, need);
  *capacity = need;
  return *buf != NULL;
}

static void configure_raw_appsrc(GstElement *appsrc, gint fps)
{
  if (!appsrc)
    return;
  g_object_set(appsrc,
      "is-live", TRUE,
      "format", GST_FORMAT_TIME,
      "block", FALSE,
      "do-timestamp", FALSE,
      NULL);
  object_set_uint_if_exists(G_OBJECT(appsrc), "max-buffers", 2);
  /* appsrc leaky-type: 2 = downstream/drop oldest on recent GStreamer. */
  object_set_int_if_exists(G_OBJECT(appsrc), "leaky-type", 2);
  {
    gchar *cap_str = g_strdup_printf(
        "video/x-raw,format=NV12,width=%d,height=%d,framerate=%d/1",
        MPP_VIDEO_WIDTH, MPP_VIDEO_HEIGHT, fps);
    GstCaps *caps = gst_caps_from_string(cap_str);
    g_object_set(appsrc, "caps", caps, NULL);
    gst_caps_unref(caps);
    g_free(cap_str);
  }
}

static void timing_store(MppH264Source *src, guint64 pts_ns, gint64 capture_us,
                         gint64 push_us, guint decode_us, guint convert_us)
{
  FrameTiming *t;

  if (!src || pts_ns == G_MAXUINT64)
    return;

  g_mutex_lock(&src->timing_lock);
  t = &src->timings[src->timing_next++ % TIMING_RING_SIZE];
  t->pts_ns = pts_ns;
  t->capture_us = capture_us;
  t->push_us = push_us;
  t->enc_in_us = 0;
  t->decode_us = decode_us;
  t->convert_us = convert_us;
  t->valid = TRUE;
  g_mutex_unlock(&src->timing_lock);
}

static void timing_mark_encoder_in(MppH264Source *src, guint64 pts_ns,
                                   gint64 enc_in_us)
{
  guint i;

  if (!src || pts_ns == G_MAXUINT64)
    return;

  g_mutex_lock(&src->timing_lock);
  for (i = 0; i < TIMING_RING_SIZE; i++) {
    FrameTiming *t = &src->timings[i];
    if (t->valid && t->pts_ns == pts_ns) {
      t->enc_in_us = enc_in_us;
      break;
    }
  }
  g_mutex_unlock(&src->timing_lock);
}

static gboolean timing_take(MppH264Source *src, guint64 pts_ns, FrameTiming *out)
{
  guint i;
  gboolean found = FALSE;

  if (!src || !out || pts_ns == G_MAXUINT64)
    return FALSE;

  g_mutex_lock(&src->timing_lock);
  for (i = 0; i < TIMING_RING_SIZE; i++) {
    FrameTiming *t = &src->timings[i];
    if (t->valid && t->pts_ns == pts_ns) {
      *out = *t;
      t->valid = FALSE;
      found = TRUE;
      break;
    }
  }
  g_mutex_unlock(&src->timing_lock);
  return found;
}

static void perf_record(MppH264Source *src, const FrameTiming *t, gint64 out_us)
{
  guint encode_us;
  guint total_us;

  if (!src || !t)
    return;

  encode_us = (t->enc_in_us > 0 && out_us > t->enc_in_us)
      ? (guint)(out_us - t->enc_in_us)
      : (t->push_us > 0 && out_us > t->push_us)
          ? (guint)(out_us - t->push_us)
          : 0;
  total_us = (t->capture_us > 0 && out_us > t->capture_us)
      ? (guint)(out_us - t->capture_us)
      : 0;

  src->perf_frames++;
  src->perf_decode_sum_us += t->decode_us;
  src->perf_convert_sum_us += t->convert_us;
  src->perf_encode_sum_us += encode_us;
  src->perf_total_sum_us += total_us;
  if (t->decode_us > src->perf_decode_max_us)
    src->perf_decode_max_us = t->decode_us;
  if (t->convert_us > src->perf_convert_max_us)
    src->perf_convert_max_us = t->convert_us;
  if (encode_us > src->perf_encode_max_us)
    src->perf_encode_max_us = encode_us;
  if (total_us > src->perf_total_max_us)
    src->perf_total_max_us = total_us;

  if (src->perf_frames >= PERF_LOG_FRAMES) {
    guint n = src->perf_frames;
    my_zlog_info("mpp_h264_source: perf avg/max jpeg_decode=%u/%u ms i420_nv12=%u/%u ms mpp_encode=%u/%u ms total=%u/%u ms",
                 (guint)(src->perf_decode_sum_us / n / 1000U),
                 src->perf_decode_max_us / 1000U,
                 (guint)(src->perf_convert_sum_us / n / 1000U),
                 src->perf_convert_max_us / 1000U,
                 (guint)(src->perf_encode_sum_us / n / 1000U),
                 src->perf_encode_max_us / 1000U,
                 (guint)(src->perf_total_sum_us / n / 1000U),
                 src->perf_total_max_us / 1000U);
    src->perf_frames = 0;
    src->perf_decode_sum_us = 0;
    src->perf_convert_sum_us = 0;
    src->perf_encode_sum_us = 0;
    src->perf_total_sum_us = 0;
    src->perf_decode_max_us = 0;
    src->perf_convert_max_us = 0;
    src->perf_encode_max_us = 0;
    src->perf_total_max_us = 0;
  }
}

static void note_slow_jpeg_decode(MppH264Source *src, guint decode_us)
{
  gint64 now_us;

  if (!src || decode_us <= MJPEG_DECODE_WARN_US)
    return;

  now_us = g_get_monotonic_time();
  src->jpeg_slow_count++;
  src->jpeg_slow_last_us = decode_us;
  if (decode_us > src->jpeg_slow_max_us)
    src->jpeg_slow_max_us = decode_us;

  if (src->jpeg_slow_last_warn_us > 0 &&
      now_us - src->jpeg_slow_last_warn_us < MJPEG_DECODE_WARN_INTERVAL_US)
    return;

  my_zlog_warn("mpp_h264_source: JPEG decode slow count=%u last=%u us max=%u us (>15ms, rate-limited 5s)",
               src->jpeg_slow_count, src->jpeg_slow_last_us,
               src->jpeg_slow_max_us);
  src->jpeg_slow_count = 0;
  src->jpeg_slow_last_us = 0;
  src->jpeg_slow_max_us = 0;
  src->jpeg_slow_last_warn_us = now_us;
}

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);
  }
}

typedef struct {
  struct jpeg_error_mgr pub;
  jmp_buf jump;
  char message[JMSG_LENGTH_MAX];
} MjpegErrorMgr;

static void mjpeg_error_exit(j_common_ptr cinfo)
{
  MjpegErrorMgr *err = (MjpegErrorMgr *)cinfo->err;
  (*cinfo->err->format_message)(cinfo, err->message);
  longjmp(err->jump, 1);
}

static gboolean decode_mjpeg_to_i420(MppH264Source *src,
                                     const guint8 *jpeg_data,
                                     size_t jpeg_size,
                                     guint *decode_us_out)
{
  struct jpeg_decompress_struct cinfo;
  MjpegErrorMgr jerr;
  gint64 start_us;
  gboolean ok = FALSE;
  guint width;
  guint height;
  guint uv_width;
  guint uv_height;
  guint i420_size;
  gint h0, v0, h1, v1, h2, v2;
  gboolean is_420;
  gboolean is_422;

  if (decode_us_out)
    *decode_us_out = 0;
  if (!src || !jpeg_data || jpeg_size == 0)
    return FALSE;

  memset(&cinfo, 0, sizeof(cinfo));
  memset(&jerr, 0, sizeof(jerr));
  cinfo.err = jpeg_std_error(&jerr.pub);
  jerr.pub.error_exit = mjpeg_error_exit;

  if (setjmp(jerr.jump)) {
    my_zlog_warn("mpp_h264_source: libjpeg-turbo decode failed: %s",
                 jerr.message[0] ? jerr.message : "unknown");
    jpeg_destroy_decompress(&cinfo);
    return FALSE;
  }

  start_us = g_get_monotonic_time();
  jpeg_create_decompress(&cinfo);
  jpeg_mem_src(&cinfo, (unsigned char *)jpeg_data, (unsigned long)jpeg_size);
  jpeg_read_header(&cinfo, TRUE);

  width = cinfo.image_width;
  height = cinfo.image_height;
  if (width != MPP_VIDEO_WIDTH || height != MPP_VIDEO_HEIGHT ||
      cinfo.num_components != 3) {
    my_zlog_warn("mpp_h264_source: unsupported MJPEG %ux%u components=%d",
                 width, height, cinfo.num_components);
    jpeg_destroy_decompress(&cinfo);
    return FALSE;
  }

  h0 = cinfo.comp_info[0].h_samp_factor;
  v0 = cinfo.comp_info[0].v_samp_factor;
  h1 = cinfo.comp_info[1].h_samp_factor;
  v1 = cinfo.comp_info[1].v_samp_factor;
  h2 = cinfo.comp_info[2].h_samp_factor;
  v2 = cinfo.comp_info[2].v_samp_factor;
  is_420 = h0 == 2 && v0 == 2 && h1 == 1 && v1 == 1 && h2 == 1 && v2 == 1;
  is_422 = h0 == 2 && v0 == 1 && h1 == 1 && v1 == 1 && h2 == 1 && v2 == 1;
  if (!is_420 && !is_422) {
    my_zlog_warn("mpp_h264_source: unsupported MJPEG subsampling Y=%dx%d Cb=%dx%d Cr=%dx%d",
                 h0, v0, h1, v1, h2, v2);
    jpeg_destroy_decompress(&cinfo);
    return FALSE;
  }

  uv_width = width / 2;
  uv_height = height / 2;
  i420_size = width * height + uv_width * uv_height * 2;
  if (!ensure_byte_buffer(&src->i420_buf, &src->i420_capacity, i420_size)) {
    jpeg_destroy_decompress(&cinfo);
    return FALSE;
  }
  if (is_422 &&
      (!ensure_byte_buffer(&src->jpeg_tmp_u, &src->jpeg_tmp_capacity,
                           uv_width * DCTSIZE) ||
       !ensure_byte_buffer(&src->jpeg_tmp_v, &src->jpeg_tmp_capacity,
                           uv_width * DCTSIZE))) {
    jpeg_destroy_decompress(&cinfo);
    return FALSE;
  }

  cinfo.raw_data_out = TRUE;
  cinfo.out_color_space = JCS_YCbCr;
  cinfo.do_fancy_upsampling = FALSE;
  cinfo.dct_method = JDCT_IFAST;
  jpeg_start_decompress(&cinfo);

  {
    guint8 *y_plane = src->i420_buf;
    guint8 *u_plane = y_plane + width * height;
    guint8 *v_plane = u_plane + uv_width * uv_height;
    const guint raw_rows = cinfo.max_v_samp_factor * DCTSIZE;
    JSAMPROW y_rows[DCTSIZE * 4];
    JSAMPROW u_rows[DCTSIZE * 4];
    JSAMPROW v_rows[DCTSIZE * 4];
    JSAMPARRAY planes[3] = {y_rows, u_rows, v_rows};

    while (cinfo.output_scanline < cinfo.output_height) {
      guint base_y = cinfo.output_scanline;
      guint i;
      JDIMENSION got;

      for (i = 0; i < raw_rows; i++) {
        guint y = base_y + i;
        if (y >= height)
          y = height - 1;
        y_rows[i] = y_plane + y * width;
      }

      if (is_420) {
        guint chroma_base = base_y / 2;
        for (i = 0; i < DCTSIZE; i++) {
          guint cy = chroma_base + i;
          if (cy >= uv_height)
            cy = uv_height - 1;
          u_rows[i] = u_plane + cy * uv_width;
          v_rows[i] = v_plane + cy * uv_width;
        }
      } else {
        for (i = 0; i < DCTSIZE; i++) {
          u_rows[i] = src->jpeg_tmp_u + i * uv_width;
          v_rows[i] = src->jpeg_tmp_v + i * uv_width;
        }
      }

      got = jpeg_read_raw_data(&cinfo, planes, raw_rows);
      if (got == 0)
        break;

      if (is_422) {
        guint chroma_base = base_y / 2;
        guint pairs = got / 2;
        for (i = 0; i < pairs && chroma_base + i < uv_height; i++) {
          guint x;
          guint8 *dst_u = u_plane + (chroma_base + i) * uv_width;
          guint8 *dst_v = v_plane + (chroma_base + i) * uv_width;
          guint8 *u0 = src->jpeg_tmp_u + (i * 2) * uv_width;
          guint8 *u1 = src->jpeg_tmp_u + (i * 2 + 1) * uv_width;
          guint8 *v0p = src->jpeg_tmp_v + (i * 2) * uv_width;
          guint8 *v1p = src->jpeg_tmp_v + (i * 2 + 1) * uv_width;
          for (x = 0; x < uv_width; x++) {
            dst_u[x] = (guint8)(((guint)u0[x] + (guint)u1[x] + 1U) >> 1);
            dst_v[x] = (guint8)(((guint)v0p[x] + (guint)v1p[x] + 1U) >> 1);
          }
        }
      }
    }
  }

  jpeg_finish_decompress(&cinfo);
  jpeg_destroy_decompress(&cinfo);
  ok = TRUE;

  if (decode_us_out)
    *decode_us_out = (guint)(g_get_monotonic_time() - start_us);
  return ok;
}

static gboolean i420_to_nv12(MppH264Source *src, guint *convert_us_out)
{
  const guint width = MPP_VIDEO_WIDTH;
  const guint height = MPP_VIDEO_HEIGHT;
  const guint y_size = width * height;
  const guint uv_width = width / 2;
  const guint uv_height = height / 2;
  const guint frame_size = y_size + y_size / 2;
  const guint8 *u_plane;
  const guint8 *v_plane;
  guint8 *uv_dst;
  gint64 start_us;
  guint row;

  if (convert_us_out)
    *convert_us_out = 0;
  if (!src || !src->i420_buf)
    return FALSE;
  if (!ensure_byte_buffer(&src->nv12_buf, &src->nv12_capacity, frame_size))
    return FALSE;

  start_us = g_get_monotonic_time();
  memcpy(src->nv12_buf, src->i420_buf, y_size);
  u_plane = src->i420_buf + y_size;
  v_plane = u_plane + uv_width * uv_height;
  uv_dst = src->nv12_buf + y_size;

  for (row = 0; row < uv_height; row++) {
    guint col;
    const guint8 *u = u_plane + row * uv_width;
    const guint8 *v = v_plane + row * uv_width;
    guint8 *dst = uv_dst + row * width;
    for (col = 0; col < uv_width; col++) {
      dst[col * 2] = u[col];
      dst[col * 2 + 1] = v[col];
    }
  }

  if (convert_us_out)
    *convert_us_out = (guint)(g_get_monotonic_time() - start_us);
  return TRUE;
}

static gpointer mjpeg_decode_thread(gpointer user_data)
{
  MppH264Source *src = user_data;
  guint64 fallback_pts = 0;
  const GstClockTime frame_duration =
      gst_util_uint64_scale_int(1, GST_SECOND, WEBRTCPUSH_H264_FPS);

  while (src && !src->stop_decode) {
    GstSample *sample;
    GstBuffer *inbuf;
    GstMapInfo inmap;
    GstBuffer *outbuf;
    GstMapInfo outmap;
    GstFlowReturn flow;
    GstClockTime pts;
    GstClockTime duration;
    gint64 capture_us;
    gint64 push_us;
    guint decode_us = 0;
    guint convert_us = 0;
    const guint nv12_size = MPP_VIDEO_WIDTH * MPP_VIDEO_HEIGHT * 3 / 2;

    if (!src->mjpeg_sink || !src->raw_appsrc) {
      g_usleep(10000);
      continue;
    }

    sample = gst_app_sink_try_pull_sample(
        GST_APP_SINK(src->mjpeg_sink), 200 * GST_MSECOND);
    if (!sample)
      continue;

    inbuf = gst_sample_get_buffer(sample);
    if (!inbuf) {
      gst_sample_unref(sample);
      continue;
    }

    pts = GST_BUFFER_PTS(inbuf);
    duration = GST_BUFFER_DURATION(inbuf);
    if (!GST_CLOCK_TIME_IS_VALID(pts)) {
      pts = fallback_pts;
      fallback_pts += frame_duration;
    }

    if (!GST_CLOCK_TIME_IS_VALID(duration))
      duration = frame_duration;
    else
      duration = frame_duration;

    capture_us = g_get_monotonic_time();
    if (!gst_buffer_map(inbuf, &inmap, GST_MAP_READ)) {
      gst_sample_unref(sample);
      continue;
    }

    if (!decode_mjpeg_to_i420(src, inmap.data, inmap.size, &decode_us)) {
      gst_buffer_unmap(inbuf, &inmap);
      gst_sample_unref(sample);
      continue;
    }
    gst_buffer_unmap(inbuf, &inmap);
    gst_sample_unref(sample);

    note_slow_jpeg_decode(src, decode_us);

    if (!i420_to_nv12(src, &convert_us))
      continue;

    outbuf = gst_buffer_new_allocate(NULL, nv12_size, NULL);
    if (!outbuf)
      continue;
    if (!gst_buffer_map(outbuf, &outmap, GST_MAP_WRITE)) {
      gst_buffer_unref(outbuf);
      continue;
    }
    memcpy(outmap.data, src->nv12_buf, nv12_size);
    gst_buffer_unmap(outbuf, &outmap);

    GST_BUFFER_PTS(outbuf) = pts;
    GST_BUFFER_DTS(outbuf) = GST_CLOCK_TIME_NONE;
    GST_BUFFER_DURATION(outbuf) = duration;

    push_us = g_get_monotonic_time();
    timing_store(src, (guint64)pts, capture_us, push_us,
                 decode_us, convert_us);
    flow = gst_app_src_push_buffer(GST_APP_SRC(src->raw_appsrc), outbuf);
    if (flow != GST_FLOW_OK && !src->stop_decode)
      my_zlog_warn("mpp_h264_source: appsrc push failed flow=%d", flow);
  }

  if (src && src->raw_appsrc)
    gst_app_src_end_of_stream(GST_APP_SRC(src->raw_appsrc));
  return 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;
  GstBuffer *buffer;

  (void)pad;
  if (!(GST_PAD_PROBE_INFO_TYPE(info) & GST_PAD_PROBE_TYPE_BUFFER))
    return GST_PAD_PROBE_OK;
  if (!src)
    return GST_PAD_PROBE_OK;

  buffer = GST_PAD_PROBE_INFO_BUFFER(info);
  if (buffer)
    timing_mark_encoder_in(src, (guint64)GST_BUFFER_PTS(buffer),
                           g_get_monotonic_time());

  if (!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;
}

static gboolean start_manual_mjpeg_pipeline(MppH264Source *src,
                                            const char *vdev,
                                            CaptureMode capture_mode,
                                            char **error_message)
{
  GstElement *cap_pipe = NULL;
  GstElement *enc_pipe = NULL;
  GstElement *vsrc = NULL;
  GstElement *caps_in = NULL;
  GstElement *jpegparse = NULL;
  GstElement *mjpeg_sink = NULL;
  GstElement *rawsrc = NULL;
  GstElement *vrate = NULL;
  GstElement *caps_nv12 = NULL;
  GstElement *q = NULL;
  GstElement *eq = NULL;
  GstElement *enc = NULL;
  GstElement *parse = NULL;
  GstElement *caps_h264 = NULL;
  GstElement *sink = NULL;
  gboolean use_mpp = FALSE;
  gint mjpeg_fps = (capture_mode == CAPTURE_MJPEG_720P24)
      ? WEBRTCPUSH_H264_FPS
      : 30;
  GstStateChangeReturn ret;

  cap_pipe = gst_pipeline_new("mjpeg-capture-pipe");
  enc_pipe = gst_pipeline_new("mpp-h264-pipe");
  vsrc = gst_element_factory_make("v4l2src", "vsrc");
  caps_in = gst_element_factory_make("capsfilter", "caps_in");
  jpegparse = gst_element_factory_make("jpegparse", "jpegparse");
  mjpeg_sink = gst_element_factory_make("appsink", "mjpeg_sink");

  rawsrc = gst_element_factory_make("appsrc", "rawsrc");
  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, FALSE, &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 (!cap_pipe || !enc_pipe || !vsrc || !caps_in || !jpegparse ||
      !mjpeg_sink || !rawsrc || !vrate || !caps_nv12 || !q || !eq ||
      !enc || !parse || !caps_h264 || !sink) {
    if (error_message)
      *error_message = g_strdup("manual MJPEG element create failed");
    goto fail;
  }

  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);
  }
  g_object_set(mjpeg_sink,
      "emit-signals", FALSE,
      "sync", FALSE,
      "max-buffers", 2,
      "drop", TRUE,
      NULL);

  configure_raw_appsrc(rawsrc, mjpeg_fps);
  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",
        MPP_VIDEO_WIDTH, 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);

  {
    GstElement *cap_elems[] = {vsrc, caps_in, jpegparse, mjpeg_sink};
    GstElement *enc_elems[] = {rawsrc, vrate, caps_nv12, q, enc, parse,
                               caps_h264, eq, sink};
    if (!link_pipeline(cap_pipe, cap_elems, G_N_ELEMENTS(cap_elems))) {
      if (error_message)
        *error_message = g_strdup("failed to link MJPEG capture chain");
      goto fail;
    }
    if (!link_pipeline(enc_pipe, enc_elems, G_N_ELEMENTS(enc_elems))) {
      if (error_message)
        *error_message = g_strdup("failed to link appsrc H264 encode chain");
      goto fail;
    }
  }

  src->manual_mjpeg = TRUE;
  src->capture_pipeline = cap_pipe;
  src->pipeline = enc_pipe;
  src->vsrc = vsrc;
  src->mjpeg_sink = mjpeg_sink;
  src->raw_appsrc = rawsrc;
  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);
    }
  }

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

  gst_pipeline_set_latency(GST_PIPELINE(enc_pipe), 0);
  gst_pipeline_set_latency(GST_PIPELINE(cap_pipe), 0);
  ret = gst_element_set_state(enc_pipe, GST_STATE_PLAYING);
  if (ret == GST_STATE_CHANGE_FAILURE) {
    if (error_message)
      *error_message = g_strdup("manual encoder pipeline failed to PLAYING");
    goto fail_started;
  }
  ret = gst_element_set_state(cap_pipe, GST_STATE_PLAYING);
  if (ret == GST_STATE_CHANGE_FAILURE) {
    if (error_message)
      *error_message = g_strdup("manual capture pipeline failed to PLAYING");
    goto fail_started;
  }

  src->decode_thread = g_thread_new("mjpeg-decode", mjpeg_decode_thread, src);

  {
    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("manual encoder warmup timeout (no frame in 8s)");
      my_zlog_error("mpp_h264_source: manual MJPEG warmup failed");
      goto fail_started;
    }
    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: 720p MJPEG@%dfps -> libjpeg-turbo(I420) -> NV12@%dfps -> mpph264enc",
               mjpeg_fps, WEBRTCPUSH_H264_FPS);
  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);
  return TRUE;

fail_started:
  if (src) {
    src->stop_decode = TRUE;
    if (src->raw_appsrc)
      gst_app_src_end_of_stream(GST_APP_SRC(src->raw_appsrc));
    if (src->capture_pipeline)
      gst_element_set_state(src->capture_pipeline, GST_STATE_NULL);
    if (src->pipeline)
      gst_element_set_state(src->pipeline, GST_STATE_NULL);
    if (src->decode_thread) {
      g_thread_join(src->decode_thread);
      src->decode_thread = NULL;
    }
    if (src->bus_watch_id) {
      g_source_remove(src->bus_watch_id);
      src->bus_watch_id = 0;
    }
    if (src->capture_bus_watch_id) {
      g_source_remove(src->capture_bus_watch_id);
      src->capture_bus_watch_id = 0;
    }
    if (src->capture_pipeline) {
      gst_object_unref(src->capture_pipeline);
      src->capture_pipeline = NULL;
    }
    if (src->pipeline) {
      gst_object_unref(src->pipeline);
      src->pipeline = NULL;
    }
    src->vsrc = NULL;
    src->mjpeg_sink = NULL;
    src->raw_appsrc = NULL;
    src->enc = NULL;
    src->parse = NULL;
    src->eq = NULL;
    src->appsink = NULL;
  }
  return FALSE;

fail:
  if (cap_pipe) {
    gst_element_set_state(cap_pipe, GST_STATE_NULL);
    gst_object_unref(cap_pipe);
  }
  if (enc_pipe) {
    gst_element_set_state(enc_pipe, GST_STATE_NULL);
    gst_object_unref(enc_pipe);
  }
  return FALSE;
}

MppH264Source *mpp_h264_source_start(const char *video_device, char **error_message)
{
  MppH264Source *src;
  GstElement *pipe, *vsrc, *caps_in = 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;
  CaptureMode capture_mode = CAPTURE_NONE;
  GstStateChangeReturn ret;

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

  src = g_new0(MppH264Source, 1);
  g_mutex_init(&src->timing_lock);

  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;
  }

  if (!use_test && capture_mode != CAPTURE_RAW_NV12_720P) {
    if (start_manual_mjpeg_pipeline(src, vdev, capture_mode, error_message)) {
      g_free(auto_v4l);
      return src;
    }
    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)");
  }

  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;
    }
  }

  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);
    if (src->manual_mjpeg && GST_CLOCK_TIME_IS_VALID(pts)) {
      FrameTiming timing;
      if (timing_take(src, (guint64)pts, &timing))
        perf_record(src, &timing, g_get_monotonic_time());
    }
    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;

  /* 编码帧之间应用动态码率，避免从 REMB/RTP 回调线程直接重配编码器。 */
  apply_pending_bitrate(src);

  /* 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);
}

static void apply_bitrate_now(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);
  }
  src->applied_bitrate_bps = bitrate_bps;
  my_zlog_info("mpp_h264_source: bitrate applied at frame boundary %u kbps",
               bitrate_bps / 1000);
}

static void apply_pending_bitrate(MppH264Source *src)
{
  gint pending;

  if (!src)
    return;
  pending = g_atomic_int_get(&src->pending_bitrate_bps);
  if (pending <= 0 ||
      !g_atomic_int_compare_and_exchange(&src->pending_bitrate_bps,
                                         pending, 0))
    return;
  if ((guint)pending == src->applied_bitrate_bps)
    return;
  apply_bitrate_now(src, (guint)pending);
}

void mpp_h264_source_set_bitrate(MppH264Source *src, guint bitrate_bps)
{
  if (!src || !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;
  g_atomic_int_set(&src->pending_bitrate_bps, (gint)bitrate_bps);
}

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;
  if (src->capture_pipeline)
    gst_element_set_state(src->capture_pipeline, GST_STATE_NULL);
  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 (src->capture_pipeline) {
    ret = gst_element_set_state(src->capture_pipeline, GST_STATE_PLAYING);
    if (ret == GST_STATE_CHANGE_FAILURE) {
      my_zlog_error("mpp_h264_source: capture 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;

  src->stop_decode = TRUE;
  if (src->raw_appsrc)
    gst_app_src_end_of_stream(GST_APP_SRC(src->raw_appsrc));
  if (src->capture_pipeline)
    gst_element_set_state(src->capture_pipeline, GST_STATE_NULL);
  if (src->pipeline)
    gst_element_set_state(src->pipeline, GST_STATE_NULL);
  if (src->decode_thread) {
    g_thread_join(src->decode_thread);
    src->decode_thread = NULL;
  }

  if (src->bus_watch_id) {
    g_source_remove(src->bus_watch_id);
    src->bus_watch_id = 0;
  }
  if (src->capture_bus_watch_id) {
    g_source_remove(src->capture_bus_watch_id);
    src->capture_bus_watch_id = 0;
  }
  if (src->capture_pipeline) {
    gst_object_unref(src->capture_pipeline);
    src->capture_pipeline = NULL;
  }
  if (src->pipeline) {
    gst_object_unref(src->pipeline);
    src->pipeline = NULL;
  }
  src->enc = NULL;
  src->parse = NULL;
  src->eq = NULL;
  src->appsink = NULL;
  src->mjpeg_sink = NULL;
  src->raw_appsrc = NULL;
  g_free(src->sprop_parameter_sets);
  g_free(src->frame_buf);
  g_free(src->i420_buf);
  g_free(src->nv12_buf);
  g_free(src->jpeg_tmp_u);
  g_free(src->jpeg_tmp_v);
  g_mutex_clear(&src->timing_lock);
  g_free(src);
}
