#include "rtc_client.h"

#include "mpp_h264_source.h"
#include "audio_source.h"
#include "audio_sink.h"
#include "volume_control.h"
#include "webrtcpush_config.h"
#include "webrtcpush_log.h"
#include "ws_signaling.h"

#include <rtc/rtc.h>

#include <glib/gstdio.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define WEBRTCPUSH_DC_AUDIO_MAX_BYTES (2U * 1024U * 1024U)
#define WEBRTCPUSH_DC_AUDIO_DIR "/tmp/webrtcpush_audio"

typedef struct {
  gchar *candidate;
  gchar *mid;
  gchar *ufrag;
} PendingRemoteIce;

typedef struct {
  gint payload_type;
  gint mline_index;
  gchar *mid;
  gchar *ice_ufrag;
} OfferVideo;

typedef struct {
  gint payload_type;
  gchar *mid;
  gboolean found;
  gint direction; /* 0=unknown 1=sendonly 2=recvonly 3=sendrecv 4=inactive */
} OfferAudio;

typedef struct {
  guint8 *data;
  gsize size;
  gboolean is_idr;
  guint64 pts_ns;
} RtcH264Frame;

typedef struct {
  GMutex lock;
  GCond not_empty;
  GQueue queue;
  guint max_depth;
  volatile gboolean stopping;
  guint dropped_total;
} RtcFrameQueue;

typedef struct {
  guint8 *data;
  gsize size;
  guint seq;
  void *client_ptr;  /* RtcClient*, 用void*避免前向声明问题 */
} DcMp3PlayTask;

typedef struct {
  AppState *app;
  GMutex lock;
  GMutex send_lock;
  GCond track_cond;
  GThread *pull_thread;
  GThread *send_thread;
  RtcFrameQueue frame_queue;
  MppH264Source *h264_source;
  gboolean stopping;
  gboolean track_open;
  gboolean need_idr;
  gboolean first_frame_sent; /* 首帧 IDR 不受 pacing flush 影响 */
  gint64 last_idr_sent_us;
  gboolean answer_sent;
  int pc;
  int track;
  guint32 rtp_timestamp;
  guint32 rtp_timestamp_base;
  guint64 source_pts_base;
  guint64 last_source_pts;
  gboolean source_pts_valid;
  guint target_bitrate;
  guint applied_mpp_bitrate;
  guint remb_ceiling;
  guint last_remb_bitrate;
  guint remb_filtered;
  guint remb_down_samples;
  gboolean initial_probe_complete;
  gint64 probe_target_reached_us;
  gint64 last_bitrate_ramp_us;
  gint64 last_mpp_bitrate_apply_us;
  gint64 last_idr_request_us;
  gint64 last_stats_log_us;
  gint64 last_remb_sane_log_us;
  gchar *remote_ice_ufrag;
  gchar *video_mid;
  gint video_mline_index;

  /* 音频 Opus track */
  AudioSource *audio_source;
  AudioSink *audio_sink;
  VolumeControl *volume_ctrl;
  int audio_track;
  int audio_payload_type;
  gboolean audio_track_open;
  gboolean audio_send_enabled;
  GThread *audio_send_thread;
  guint32 audio_rtp_timestamp;
  guint64 audio_pts_base;
  guint64 audio_last_pts;
  gboolean audio_pts_valid;

  /* DataChannel: 前端 MP3 分片 + 右上角 Mbps */
  int data_channel;
  GByteArray *dc_audio_buf;
  guint dc_audio_seq;
  /* 排队播放: 多片段连续播放不间隔 */
  GAsyncQueue *dc_play_queue;     /* DcMp3PlayTask队列 */
  GThread *dc_play_thread;        /* 常驻播放线程 */
  volatile gboolean dc_play_quit;

  GQueue *pending_local_ice;
  GQueue *pending_remote_ice;
} RtcClient;

static const char *s_ice_servers[] = {
    "stun:8.129.3.163:18084",
    "stun:stun.l.google.com:19302",
    "turn:coturn:coturn@8.129.3.163:18084?transport=udp",
    "turn:coturn:coturn@8.129.3.163:18084?transport=tcp",
};

static RtcClient *client_from_app(AppState *app)
{
  return app ? (RtcClient *)app->rtc_client : NULL;
}

static RtcH264Frame *rtc_h264_frame_new(const guint8 *data, gsize size,
                                        gboolean is_idr, guint64 pts_ns)
{
  RtcH264Frame *frame = g_new(RtcH264Frame, 1);

  frame->data = g_memdup2(data, size);
  frame->size = size;
  frame->is_idr = is_idr;
  frame->pts_ns = pts_ns;
  return frame;
}

static void rtc_h264_frame_free(RtcH264Frame *frame)
{
  if (!frame)
    return;
  g_free(frame->data);
  g_free(frame);
}

static void frame_queue_init(RtcFrameQueue *q, guint max_depth)
{
  g_mutex_init(&q->lock);
  g_cond_init(&q->not_empty);
  g_queue_init(&q->queue);
  q->max_depth = max_depth;
  q->stopping = FALSE;
  q->dropped_total = 0;
}

static void frame_queue_fini(RtcFrameQueue *q)
{
  RtcH264Frame *frame;

  g_mutex_lock(&q->lock);
  while ((frame = g_queue_pop_head(&q->queue)) != NULL)
    rtc_h264_frame_free(frame);
  g_mutex_unlock(&q->lock);
  g_queue_clear(&q->queue);
  g_cond_clear(&q->not_empty);
  g_mutex_clear(&q->lock);
}

static void frame_queue_stop(RtcFrameQueue *q)
{
  g_mutex_lock(&q->lock);
  q->stopping = TRUE;
  g_cond_broadcast(&q->not_empty);
  g_mutex_unlock(&q->lock);
}

static guint frame_queue_clear(RtcClient *client)
{
  RtcFrameQueue *q = &client->frame_queue;
  RtcH264Frame *frame;
  guint dropped = 0;

  if (!client)
    return 0;

  g_mutex_lock(&q->lock);
  while ((frame = g_queue_pop_head(&q->queue)) != NULL) {
    dropped++;
    q->dropped_total++;
    rtc_h264_frame_free(frame);
  }
  g_mutex_unlock(&q->lock);
  return dropped;
}

static void frame_queue_push(RtcClient *client, RtcH264Frame *frame)
{
  RtcFrameQueue *q = &client->frame_queue;
  RtcH264Frame *old;
  gboolean overflowed = FALSE;
  gboolean request_resync = FALSE;
  guint dropped = 0;

  g_mutex_lock(&q->lock);
  if (q->stopping) {
    g_mutex_unlock(&q->lock);
    rtc_h264_frame_free(frame);
    return;
  }
  if (g_queue_get_length(&q->queue) >= q->max_depth) {
    overflowed = TRUE;
    while ((old = g_queue_pop_head(&q->queue)) != NULL) {
      dropped++;
      q->dropped_total++;
      rtc_h264_frame_free(old);
    }
  }
  if (overflowed && !frame->is_idr) {
    dropped++;
    q->dropped_total++;
    rtc_h264_frame_free(frame);
    request_resync = TRUE;
  } else {
    g_queue_push_tail(&q->queue, frame);
    g_cond_signal(&q->not_empty);
  }
  g_mutex_unlock(&q->lock);

  /* Never continue a predictive chain after dropping an encoded P frame. */
  if (request_resync) {
    g_mutex_lock(&client->lock);
    client->need_idr = TRUE;
    client->source_pts_valid = FALSE;
    client->last_idr_request_us = g_get_monotonic_time();
    g_mutex_unlock(&client->lock);
    my_zlog_warn("libdatachannel: encoded-frame queue overflow; dropped=%u frame(s), request clean IDR",
                 dropped);
  }
}

static RtcH264Frame *frame_queue_pop(RtcClient *client, gint timeout_ms)
{
  RtcFrameQueue *q = &client->frame_queue;
  RtcH264Frame *frame = NULL;
  gint64 end_us = g_get_monotonic_time() + (gint64)timeout_ms * 1000;

  g_mutex_lock(&q->lock);
  while (g_queue_is_empty(&q->queue) && !q->stopping) {
    if (!g_cond_wait_until(&q->not_empty, &q->lock, end_us))
      break;
  }
  if (!g_queue_is_empty(&q->queue))
    frame = g_queue_pop_head(&q->queue);
  g_mutex_unlock(&q->lock);
  return frame;
}

static void free_pending_remote_ice(gpointer data)
{
  PendingRemoteIce *ice = data;
  if (!ice)
    return;
  g_free(ice->candidate);
  g_free(ice->mid);
  g_free(ice->ufrag);
  g_free(ice);
}

static gchar *member_id_copy(AppState *app)
{
  gchar *member_id;
  app_state_lock(app);
  member_id = g_strdup(app->member_id);
  app_state_unlock(app);
  return member_id;
}

static gchar *build_description_json(AppState *app, const char *type,
                                     const char *sdp)
{
  JsonBuilder *builder;
  JsonGenerator *generator;
  JsonNode *root;
  gchar *json;
  gchar *member_id = member_id_copy(app);

  if (!member_id || !sdp) {
    g_free(member_id);
    return NULL;
  }

  builder = json_builder_new();
  json_builder_begin_object(builder);
  json_builder_set_member_name(builder, "memberId");
  json_builder_add_string_value(builder, member_id);
  json_builder_set_member_name(builder, "description");
  json_builder_begin_object(builder);
  json_builder_set_member_name(builder, "type");
  json_builder_add_string_value(builder, type);
  json_builder_set_member_name(builder, "sdp");
  json_builder_add_string_value(builder, sdp);
  json_builder_end_object(builder);
  json_builder_end_object(builder);

  generator = json_generator_new();
  root = json_builder_get_root(builder);
  json_generator_set_root(generator, root);
  json = json_generator_to_data(generator, NULL);
  json_node_unref(root);
  g_object_unref(generator);
  g_object_unref(builder);
  g_free(member_id);
  return json;
}

static gchar *build_candidate_json(RtcClient *client, const char *candidate,
                                   const char *mid)
{
  JsonBuilder *builder;
  JsonGenerator *generator;
  JsonNode *root;
  gchar *json;
  gchar *member_id = member_id_copy(client->app);
  gint mline;

  if (!member_id || !candidate) {
    g_free(member_id);
    return NULL;
  }
  g_mutex_lock(&client->lock);
  mline = client->video_mline_index;
  g_mutex_unlock(&client->lock);

  builder = json_builder_new();
  json_builder_begin_object(builder);
  json_builder_set_member_name(builder, "memberId");
  json_builder_add_string_value(builder, member_id);
  json_builder_set_member_name(builder, "candidate");
  json_builder_begin_object(builder);
  json_builder_set_member_name(builder, "candidate");
  json_builder_add_string_value(builder, candidate);
  json_builder_set_member_name(builder, "sdpMLineIndex");
  json_builder_add_int_value(builder, mline);
  json_builder_set_member_name(builder, "sdpMid");
  json_builder_add_string_value(builder, mid ? mid : "0");
  json_builder_end_object(builder);
  json_builder_end_object(builder);

  generator = json_generator_new();
  root = json_builder_get_root(builder);
  json_generator_set_root(generator, root);
  json = json_generator_to_data(generator, NULL);
  json_node_unref(root);
  g_object_unref(generator);
  g_object_unref(builder);
  g_free(member_id);
  return json;
}

static void flush_local_ice(RtcClient *client)
{
  for (;;) {
    gchar *json;
    g_mutex_lock(&client->lock);
    json = g_queue_pop_head(client->pending_local_ice);
    g_mutex_unlock(&client->lock);
    if (!json)
      break;
    ws_send_text(client->app, json);
    g_free(json);
  }
  ws_signaling_service_burst(1);
}

static gchar *sdp_replace_first(const gchar *haystack, const gchar *needle,
                                const gchar *replacement)
{
  const gchar *pos;

  if (!haystack || !needle || !replacement)
    return g_strdup(haystack);
  pos = strstr(haystack, needle);
  if (!pos)
    return g_strdup(haystack);
  {
    gsize prefix = (gsize)(pos - haystack);
    GString *out = g_string_new_len(haystack, prefix);
    g_string_append(out, replacement);
    g_string_append(out, pos + strlen(needle));
    return g_string_free(out, FALSE);
  }
}

static gchar *sdp_patch_h264_answer(const gchar *sdp, const gchar *sprop)
{
  gchar *out;
  gchar *needle;
  gchar *insert;

  if (!sdp)
    return NULL;
  out = g_strdup(sdp);
  if (!sprop || !sprop[0] || strstr(out, "sprop-parameter-sets"))
    return out;

  needle = g_strdup_printf("profile-level-id=%s", WEBRTCPUSH_H264_PROFILE_LEVEL_ID);
  insert = g_strdup_printf("profile-level-id=%s;sprop-parameter-sets=%s",
                           WEBRTCPUSH_H264_PROFILE_LEVEL_ID, sprop);
  if (strstr(out, needle)) {
    gchar *replaced = sdp_replace_first(out, needle, insert);
    g_free(out);
    out = replaced;
    my_zlog_info("libdatachannel: SDP answer patched with sprop-parameter-sets");
  } else {
    my_zlog_warn("libdatachannel: SDP answer missing profile-level-id, sprop not patched");
  }
  g_free(needle);
  g_free(insert);
  return out;
}

static gchar *sdp_ensure_msid_semantic(const gchar *sdp)
{
  const gchar *mline;

  if (!sdp || strstr(sdp, "a=msid-semantic:"))
    return g_strdup(sdp);
  mline = strstr(sdp, "m=");
  if (!mline)
    return g_strdup(sdp);
  {
    GString *out = g_string_new_len(sdp, (gsize)(mline - sdp));
    g_string_append(out, "a=msid-semantic: WMS car-stream\r\n");
    g_string_append(out, mline);
    my_zlog_info("libdatachannel: SDP answer patched with msid-semantic");
    return g_string_free(out, FALSE);
  }
}

static gchar *build_h264_profile_fmtp(RtcClient *client)
{
  const gchar *sprop = mpp_h264_source_get_sprop(client->h264_source);

  if (sprop && sprop[0])
    return g_strdup_printf(
        "profile-level-id=%s;packetization-mode=1;level-asymmetry-allowed=1;"
        "sprop-parameter-sets=%s",
        WEBRTCPUSH_H264_PROFILE_LEVEL_ID, sprop);
  return g_strdup_printf(
      "profile-level-id=%s;packetization-mode=1;level-asymmetry-allowed=1",
      WEBRTCPUSH_H264_PROFILE_LEVEL_ID);
}

static void RTC_API on_local_description(int pc, const char *sdp,
                                         const char *type, void *ptr)
{
  RtcClient *client = ptr;
  gchar *json;
  gchar *patched = NULL;
  const gchar *sprop;
  (void)pc;

  if (!client || !sdp || g_strcmp0(type, "answer") != 0)
    return;
  sprop = mpp_h264_source_get_sprop(client->h264_source);
  patched = sdp_patch_h264_answer(sdp, sprop);
  {
    gchar *with_msid = sdp_ensure_msid_semantic(patched ? patched : sdp);
    g_free(patched);
    patched = with_msid;
  }
  my_zlog_info("libdatachannel: SDP answer:\n%s", patched ? patched : sdp);
  json = build_description_json(client->app, "answer", patched ? patched : sdp);
  g_free(patched);
  if (!json) {
    my_zlog_error("libdatachannel: cannot send answer without memberId");
    return;
  }
  ws_send_text(client->app, json);
  g_free(json);
  g_mutex_lock(&client->lock);
  client->answer_sent = TRUE;
  g_mutex_unlock(&client->lock);
  flush_local_ice(client);
  my_zlog_info("libdatachannel: SDP answer sent");
}

static void RTC_API on_local_candidate(int pc, const char *candidate,
                                       const char *mid, void *ptr)
{
  RtcClient *client = ptr;
  gchar *json;
  gboolean send_now;
  (void)pc;

  if (!client || !candidate || !candidate[0])
    return;
  my_zlog_info("libdatachannel: local candidate mid=%s: %s", mid ? mid : "?", candidate);
  json = build_candidate_json(client, candidate, mid);
  if (!json)
    return;
  g_mutex_lock(&client->lock);
  send_now = client->answer_sent;
  if (!send_now)
    g_queue_push_tail(client->pending_local_ice, json);
  g_mutex_unlock(&client->lock);
  if (send_now) {
    ws_send_text(client->app, json);
    g_free(json);
    ws_signaling_service_burst(1);
  }
}

static const char *rtc_state_name(rtcState state)
{
  switch (state) {
  case RTC_NEW: return "new";
  case RTC_CONNECTING: return "connecting";
  case RTC_CONNECTED: return "connected";
  case RTC_DISCONNECTED: return "disconnected";
  case RTC_FAILED: return "failed";
  case RTC_CLOSED: return "closed";
  default: return "unknown";
  }
}

static void signal_track_ready(RtcClient *client)
{
  if (!client)
    return;
  g_cond_broadcast(&client->track_cond);
}

static gboolean request_idr(RtcClient *client, gboolean bypass_throttle)
{
  gint64 now;

  if (!client)
    return FALSE;
  now = g_get_monotonic_time();
  g_mutex_lock(&client->lock);
  if (client->need_idr) {
    g_mutex_unlock(&client->lock);
    return FALSE;
  }
  if (!bypass_throttle && client->last_idr_request_us > 0 &&
      now - client->last_idr_request_us <
          (gint64)WEBRTCPUSH_PLI_IDR_THROTTLE_MS * 1000) {
    g_mutex_unlock(&client->lock);
    return FALSE;
  }
  client->need_idr = TRUE;
  client->last_idr_request_us = now;
  g_mutex_unlock(&client->lock);
  return TRUE;
}

static void log_selected_ice_path(int pc)
{
  char local_candidate[512] = {0};
  char remote_candidate[512] = {0};
  char local_address[128] = {0};
  char remote_address[128] = {0};
  int pair_ret;
  int local_ret;
  int remote_ret;

  if (pc < 0)
    return;
  pair_ret = rtcGetSelectedCandidatePair(
      pc, local_candidate, (int)sizeof(local_candidate),
      remote_candidate, (int)sizeof(remote_candidate));
  local_ret = rtcGetLocalAddress(pc, local_address, (int)sizeof(local_address));
  remote_ret = rtcGetRemoteAddress(pc, remote_address, (int)sizeof(remote_address));
  if (pair_ret >= 0) {
    my_zlog_info("libdatachannel: ICE selected local=[%s] remote=[%s] socket=%s -> %s",
                 local_candidate, remote_candidate,
                 local_ret >= 0 ? local_address : "?",
                 remote_ret >= 0 ? remote_address : "?");
  } else {
    my_zlog_warn("libdatachannel: connected but selected ICE pair unavailable (%d)",
                 pair_ret);
  }
}

static void RTC_API on_state_change(int pc, rtcState state, void *ptr)
{
  RtcClient *client = ptr;
  my_zlog_info("libdatachannel: peer state=%s", rtc_state_name(state));
  if (!client)
    return;
  if (state == RTC_CONNECTED) {
    log_selected_ice_path(pc);
    g_mutex_lock(&client->lock);
    client->source_pts_valid = FALSE;
    g_mutex_unlock(&client->lock);
    request_idr(client, TRUE);
  }
}

/* 常驻播放线程: 从队列取任务连续播放, 队列空时释放ALSA锁 */
static gpointer dc_play_queue_thread(gpointer data)
{
  RtcClient *client = data;
  gboolean holds_lock = FALSE;
  int idle_count = 0;

  while (!client->dc_play_quit) {
    /* 等300ms看有没有新任务, 有就播, 连续3次空就释放锁 */
    DcMp3PlayTask *task = g_async_queue_timeout_pop(client->dc_play_queue, 300000);
    if (!task) {
      /* 队列空 */
      if (holds_lock) {
        idle_count++;
        if (idle_count >= 3) {
          /* 约1s无新任务, 释放ALSA锁让本地音频能播 */
          audio_sink_unlock_alsa();
          holds_lock = FALSE;
          idle_count = 0;
        }
      }
      continue;
    }
    idle_count = 0;

    if (!task->data || task->size == 0) {
      g_free(task->data);
      g_free(task);
      continue;
    }

    /* 首次或有任务时获取锁, 播放期间不释放 */
    if (!holds_lock) {
      RtcClient *rc = (RtcClient*)task->client_ptr;
      audio_sink_interrupt(rc ? rc->audio_sink : NULL);
      audio_sink_lock_alsa();
      holds_lock = TRUE;
    }

    /* 写文件 + playbin播放 */
    gchar *path = g_strdup_printf("%s/phone_%u.bin", WEBRTCPUSH_DC_AUDIO_DIR, task->seq);
    GError *err = NULL;
    if (g_mkdir_with_parents(WEBRTCPUSH_DC_AUDIO_DIR, 0700) != 0 ||
        !g_file_set_contents(path, (const gchar*)task->data, (gssize)task->size, &err)) {
      my_zlog_warn("libdatachannel: save phone audio failed: %s", err ? err->message : "?");
      if (err) g_error_free(err);
      g_free(path);
      g_free(task->data);
      g_free(task);
      continue;
    }

    gchar *quoted = g_shell_quote(path);
    gchar *cmd = g_strdup_printf(
        /*
         * 修复两个问题:
         *   1) sync=false 会让 playbin 在 EOS 时立即关闭 ALSA 设备, ring buffer
         *      尾部尚未播完就被 snd_pcm_drop 丢弃, 表现为"只播放前面部分"。
         *      改 sync=true 让 sink 按时钟节拍播放并在 EOS 时 drain 完整音频。
         *   2) USB 声卡(plughw:2,0)播单声道会触发
         *      gst_audio_ring_buffer_set_channel_positions 的 "should not be reached"
         *      CRITICAL; 用 audioconvert 上混到 2 声道规避。
         */
        "timeout 20s gst-launch-1.0 -q playbin uri=file://%s audio-sink=\"audioconvert ! audioresample ! audio/x-raw,channels=2 ! alsasink device=%s sync=true\" video-sink=fakesink",
        quoted, WEBRTCPUSH_AUDIO_PLAYBACK_DEVICE);
    int status = -1;
    gchar *child_out = NULL, *child_err = NULL;
    my_zlog_info("libdatachannel: playing %u bytes (seq=%u)", (guint)task->size, task->seq);
    /* capture & discard child stdout/stderr: suppress rga_api version + gst progress noise */
    if (!g_spawn_command_line_sync(cmd, &child_out, &child_err, &status, &err)) {
      my_zlog_warn("libdatachannel: play phone audio failed: %s", err ? err->message : "?");
      if (err) g_error_free(err);
    } else if (status == 0) {
      my_zlog_info("libdatachannel: phone audio played (%u bytes)", (guint)task->size);
    } else {
      my_zlog_warn("libdatachannel: phone audio playback exit=%d (%u bytes)",
                   status, (guint)task->size);
    }
    g_remove(path);
    g_free(child_out);
    g_free(child_err);
    g_free(cmd);
    g_free(quoted);
    g_free(path);
    g_free(task->data);
    g_free(task);
  }

  /* 退出前释放锁 */
  if (holds_lock)
    audio_sink_unlock_alsa();
  return NULL;
}

static void start_dc_mp3_playback(RtcClient *client, guint8 *data, gsize size)
{
  DcMp3PlayTask *task;

  if (!client || !data || size == 0)
    return;
  /* 首次播放时创建队列+线程 */
  if (!client->dc_play_queue) {
    client->dc_play_queue = g_async_queue_new();
    client->dc_play_quit = FALSE;
    client->dc_play_thread = g_thread_new("dc-play-q", dc_play_queue_thread, client);
  }
  task = g_new0(DcMp3PlayTask, 1);
  task->data = data;
  task->size = size;
  task->seq = ++client->dc_audio_seq;
  task->client_ptr = client;
  g_async_queue_push(client->dc_play_queue, task);
}

static gboolean dc_message_is_eof(const char *message, int size)
{
  if (!message)
    return FALSE;
  if (size < 0)
    return strcmp(message, "EOF") == 0;
  return size == 3 && memcmp(message, "EOF", 3) == 0;
}

static void RTC_API on_data_channel_message(int dc, const char *message, int size, void *ptr)
{
  RtcClient *client = ptr;
  guint8 *play_data = NULL;
  gsize play_size = 0;

  (void)dc;
  if (!client || !message)
    return;

  if (dc_message_is_eof(message, size)) {
    g_mutex_lock(&client->lock);
    if (client->dc_audio_buf && client->dc_audio_buf->len > 0) {
      play_size = client->dc_audio_buf->len;
      play_data = g_memdup2(client->dc_audio_buf->data, play_size);
      g_byte_array_set_size(client->dc_audio_buf, 0);
    }
    g_mutex_unlock(&client->lock);

    if (play_data) {
      my_zlog_info("libdatachannel: DataChannel MP3 EOF, playback %u bytes", (guint)play_size);
      start_dc_mp3_playback(client, play_data, play_size);
    } else {
      my_zlog_warn("libdatachannel: DataChannel MP3 EOF with empty buffer");
    }
    return;
  }

  if (size <= 0)
    return;

  g_mutex_lock(&client->lock);
  if (!client->dc_audio_buf)
    client->dc_audio_buf = g_byte_array_new();
  if (client->dc_audio_buf->len + (guint)size > WEBRTCPUSH_DC_AUDIO_MAX_BYTES) {
    my_zlog_warn("libdatachannel: DataChannel MP3 too large (%u + %d), drop buffer",
                 client->dc_audio_buf->len, size);
    g_byte_array_set_size(client->dc_audio_buf, 0);
  } else {
    g_byte_array_append(client->dc_audio_buf, (const guint8 *)message, (guint)size);
  }
  g_mutex_unlock(&client->lock);
}

static void send_datachannel_mbps(RtcClient *client, guint actual_kbps)
{
  int dc;
  guint target_bitrate;
  char msg[32];
  double mbps;

  if (!client)
    return;
  g_mutex_lock(&client->lock);
  dc = client->data_channel;
  target_bitrate = client->target_bitrate;
  g_mutex_unlock(&client->lock);
  if (dc < 0 || !rtcIsOpen(dc))
    return;
  mbps = actual_kbps > 0 ? ((double)actual_kbps / 1000.0) :
      ((double)target_bitrate / 1000000.0);
  snprintf(msg, sizeof(msg), "%.2f", mbps);
  rtcSendMessage(dc, msg, -1);
}

static void RTC_API on_data_channel_open(int dc, void *ptr)
{
  RtcClient *client = ptr;
  char label[64] = {0};
  rtcGetDataChannelLabel(dc, label, (int)sizeof(label));
  my_zlog_info("libdatachannel: DataChannel open label=%s", label[0] ? label : "?");
  send_datachannel_mbps(client, 0);
}

static void RTC_API on_data_channel_closed(int dc, void *ptr)
{
  RtcClient *client = ptr;
  if (!client)
    return;
  g_mutex_lock(&client->lock);
  if (client->data_channel == dc)
    client->data_channel = -1;
  g_mutex_unlock(&client->lock);
  my_zlog_info("libdatachannel: DataChannel closed");
}

static void RTC_API on_data_channel_error(int dc, const char *error, void *ptr)
{
  (void)dc;
  (void)ptr;
  my_zlog_warn("libdatachannel: DataChannel error: %s", error ? error : "?");
}

static void setup_data_channel_callbacks(RtcClient *client, int dc)
{
  if (!client || dc < 0)
    return;
  rtcSetUserPointer(dc, client);
  rtcSetOpenCallback(dc, on_data_channel_open);
  rtcSetClosedCallback(dc, on_data_channel_closed);
  rtcSetErrorCallback(dc, on_data_channel_error);
  rtcSetMessageCallback(dc, on_data_channel_message);
}

static void RTC_API on_incoming_data_channel(int pc, int dc, void *ptr)
{
  RtcClient *client = ptr;
  char label[64] = {0};
  (void)pc;
  rtcGetDataChannelLabel(dc, label, (int)sizeof(label));
  my_zlog_info("libdatachannel: incoming DataChannel label=%s", label[0] ? label : "?");
  setup_data_channel_callbacks(client, dc);
}

static void RTC_API on_track_open(int track, void *ptr)
{
  RtcClient *client = ptr;
  gboolean video_open = FALSE;
  gboolean audio_open = FALSE;

  if (!client)
    return;
  g_mutex_lock(&client->lock);
  if (client->track == track) {
    client->track_open = TRUE;
    client->source_pts_valid = FALSE;
    client->first_frame_sent = FALSE;
    client->last_idr_sent_us = 0;
    video_open = TRUE;
  }
  if (client->audio_track == track) {
    client->audio_track_open = TRUE;
    client->audio_pts_valid = FALSE;
    audio_open = TRUE;
  }
  g_mutex_unlock(&client->lock);

  if (video_open) {
    frame_queue_clear(client);
    request_idr(client, TRUE);
    my_zlog_info("libdatachannel: H264 track open, requesting IDR");
  } else if (audio_open) {
    my_zlog_info("libdatachannel: Opus track open");
  }
  signal_track_ready(client);
}

static void RTC_API on_track_closed(int track, void *ptr)
{
  RtcClient *client = ptr;
  gboolean video_closed = FALSE;
  gboolean audio_closed = FALSE;

  if (!client)
    return;
  g_mutex_lock(&client->lock);
  if (client->track == track) {
    client->track_open = FALSE;
    client->first_frame_sent = FALSE;
    client->last_idr_sent_us = 0;
    video_closed = TRUE;
  }
  if (client->audio_track == track) {
    client->audio_track_open = FALSE;
    audio_closed = TRUE;
  }
  g_mutex_unlock(&client->lock);
  if (video_closed)
    frame_queue_clear(client);
  signal_track_ready(client);
  my_zlog_warn("libdatachannel: track closed (video=%d audio=%d)",
               video_closed ? 1 : 0, audio_closed ? 1 : 0);
}

static void RTC_API on_track_error(int track, const char *error, void *ptr)
{
  (void)track;
  (void)ptr;
  my_zlog_error("libdatachannel track: %s", error ? error : "unknown error");
}

static void RTC_API on_pli(int track, void *ptr)
{
  RtcClient *client = ptr;
  gboolean requested;
  int dropped = 0;
  guint dropped_frames = 0;
  if (!client || client->track != track)
    return;
  requested = request_idr(client, FALSE);
  if (!requested) {
    my_zlog_debug("libdatachannel: PLI ignored, IDR already pending or throttled");
    return;
  }
  g_mutex_lock(&client->lock);
  client->source_pts_valid = FALSE;
  g_mutex_unlock(&client->lock);
  g_mutex_lock(&client->send_lock);
  dropped = rtcClearPacingQueue(track);
  g_mutex_unlock(&client->send_lock);
  dropped_frames = frame_queue_clear(client);
  my_zlog_info("libdatachannel: RTCP PLI -> MPP IDR; dropped %d RTP packet(s), %u encoded frame(s)",
               dropped > 0 ? dropped : 0, dropped_frames);
}

static guint clamp_bitrate(guint bitrate)
{
  if (bitrate < WEBRTCPUSH_MIN_BITRATE)
    return WEBRTCPUSH_MIN_BITRATE;
  if (bitrate > WEBRTCPUSH_MAX_BITRATE)
    return WEBRTCPUSH_MAX_BITRATE;
  return bitrate;
}

static guint pacing_bitrate_for_encoder(guint encoder_bitrate)
{
  guint64 pacing = ((guint64)clamp_bitrate(encoder_bitrate) *
                    WEBRTCPUSH_PACING_HEADROOM_PERCENT) / 100;
  if (pacing > WEBRTCPUSH_PACING_MAX_BITRATE)
    pacing = WEBRTCPUSH_PACING_MAX_BITRATE;
  return (guint)pacing;
}

static gboolean pacing_queue_snapshot(RtcClient *client, int track,
                                      guint *packets_out, guint *bytes_out,
                                      guint *delay_ms_out)
{
  guint packets = 0;
  guint bytes = 0;
  guint target;
  guint pacing_bps;
  int ret = -1;

  if (!client || track < 0)
    return FALSE;
  g_mutex_lock(&client->send_lock);
  if (client->track == track)
    ret = rtcGetPacingQueueStats(track, &packets, &bytes);
  g_mutex_unlock(&client->send_lock);
  if (ret < 0)
    return FALSE;

  g_mutex_lock(&client->lock);
  target = client->target_bitrate;
  g_mutex_unlock(&client->lock);
  pacing_bps = pacing_bitrate_for_encoder(target);
  if (packets_out)
    *packets_out = packets;
  if (bytes_out)
    *bytes_out = bytes;
  if (delay_ms_out) {
    *delay_ms_out = pacing_bps > 0
        ? (guint)(((guint64)bytes * 8U * 1000U + pacing_bps - 1U) /
                  pacing_bps)
        : 0;
  }
  return TRUE;
}

static int clear_stale_pacing(RtcClient *client, int track,
                              guint *dropped_frames_out)
{
  int dropped = 0;
  guint dropped_frames;

  g_mutex_lock(&client->lock);
  client->source_pts_valid = FALSE;
  g_mutex_unlock(&client->lock);
  g_mutex_lock(&client->send_lock);
  if (client->track == track)
    dropped = rtcClearPacingQueue(track);
  g_mutex_unlock(&client->send_lock);
  dropped_frames = frame_queue_clear(client);
  if (dropped_frames_out)
    *dropped_frames_out = dropped_frames;
  return dropped;
}

static guint bitrate_delta(guint a, guint b)
{
  return a > b ? a - b : b - a;
}

static void apply_bitrate(RtcClient *client, int track, guint bitrate)
{
  guint b;
  guint old_b;
  guint old_mpp_b;
  guint mpp_apply_b = 0;
  guint mpp_delta;
  gint64 now_us;
  gint64 last_mpp_apply_us;
  gboolean apply_mpp = FALSE;

  if (!client)
    return;
  b = clamp_bitrate(bitrate);
  now_us = g_get_monotonic_time();

  g_mutex_lock(&client->lock);
  old_b = client->target_bitrate;
  old_mpp_b = client->applied_mpp_bitrate;
  last_mpp_apply_us = client->last_mpp_bitrate_apply_us;
  if (old_mpp_b == 0)
    old_mpp_b = old_b ? old_b : b;
  client->target_bitrate = b;
  mpp_delta = bitrate_delta(old_mpp_b, b);
  if ((guint64)mpp_delta * 100U >=
          (guint64)old_mpp_b * WEBRTCPUSH_MPP_RECONFIG_MIN_DELTA_PERCENT &&
      now_us - last_mpp_apply_us >=
          (gint64)WEBRTCPUSH_MPP_RECONFIG_MIN_INTERVAL_MS * 1000) {
    apply_mpp = TRUE;
    mpp_apply_b = b;
    if (b > old_mpp_b) {
      guint max_up = (guint)(((guint64)old_mpp_b *
          (100U + WEBRTCPUSH_MPP_RECONFIG_MAX_STEP_PERCENT)) / 100U);
      if (mpp_apply_b > max_up)
        mpp_apply_b = max_up;
    } else {
      guint min_down = (guint)(((guint64)old_mpp_b *
          (100U - WEBRTCPUSH_MPP_RECONFIG_MAX_STEP_PERCENT)) / 100U);
      if (mpp_apply_b < min_down)
        mpp_apply_b = min_down;
    }
    mpp_apply_b = clamp_bitrate(mpp_apply_b);
    client->applied_mpp_bitrate = mpp_apply_b;
    client->last_mpp_bitrate_apply_us = now_us;
  }
  g_mutex_unlock(&client->lock);

  if (old_b == b)
    return;

  if (track >= 0) {
    g_mutex_lock(&client->send_lock);
    if (client->track == track)
      rtcSetPacingBitrate(track, pacing_bitrate_for_encoder(b));
    g_mutex_unlock(&client->send_lock);
  }
  if (apply_mpp) {
    /* 这里只投递原子目标；真正的 mpph264enc 属性更新在拉帧边界执行。 */
    mpp_h264_source_set_bitrate(client->h264_source, mpp_apply_b);
    my_zlog_info("libdatachannel: bitrate control -> target=%u->%u kbps, MPP queued=%u->%u kbps pacing=%u kbps",
                 old_b / 1000, b / 1000, old_mpp_b / 1000, mpp_apply_b / 1000,
                 pacing_bitrate_for_encoder(b) / 1000);
  } else {
    my_zlog_info("libdatachannel: bitrate control -> target=%u->%u kbps, MPP hold=%u kbps pacing=%u kbps",
                 old_b / 1000, b / 1000, old_mpp_b / 1000,
                 pacing_bitrate_for_encoder(b) / 1000);
  }
}

/* Public Internet: climb slowly so a transient REMB spike cannot build latency. */
static void ramp_bitrate_toward_ceiling(RtcClient *client, int track)
{
  guint cur;
  guint tgt;
  guint next;
  guint diff;
  guint step;

  if (!client || track < 0)
    return;

  g_mutex_lock(&client->lock);
  cur = client->target_bitrate;
  tgt = client->remb_ceiling;
  g_mutex_unlock(&client->lock);

  if (!cur || !tgt || cur == tgt)
    return;

  if (cur > tgt)
    return; /* confirmed decreases are applied by on_remb() */

  diff = tgt - cur;
  step = (guint)(((guint64)cur * WEBRTCPUSH_BITRATE_RAMP_UP_PERCENT) / 100U);
  if (step < WEBRTCPUSH_BITRATE_RAMP_UP_STEP_MIN_BPS)
    step = WEBRTCPUSH_BITRATE_RAMP_UP_STEP_MIN_BPS;
  if (step > WEBRTCPUSH_BITRATE_RAMP_UP_STEP_MAX_BPS)
    step = WEBRTCPUSH_BITRATE_RAMP_UP_STEP_MAX_BPS;
  if (step > diff)
    step = diff;
  next = cur + step;

  if (next == cur)
    return;

  apply_bitrate(client, track, next);
  my_zlog_info("libdatachannel: bitrate ramp %u -> %u kbps (REMB ceiling %u kbps)",
               cur / 1000, next / 1000, tgt / 1000);
}

static guint bitrate_step_down_toward(guint current, guint ceiling)
{
  guint diff;
  guint step;

  if (ceiling >= current)
    return current;

  diff = current - ceiling;
  step = (guint)(((guint64)current * WEBRTCPUSH_REMB_DOWN_STEP_PERCENT) / 100U);
  if (step < WEBRTCPUSH_REMB_DOWN_STEP_MIN_BPS)
    step = WEBRTCPUSH_REMB_DOWN_STEP_MIN_BPS;
  if (step > WEBRTCPUSH_REMB_DOWN_STEP_MAX_BPS)
    step = WEBRTCPUSH_REMB_DOWN_STEP_MAX_BPS;
  if (step > diff)
    step = diff;
  return current - step;
}

static void RTC_API on_remb(int track, unsigned int bitrate, void *ptr)
{
  RtcClient *client = ptr;
  guint current;
  guint usable;
  guint estimate_bitrate;
  guint filtered;
  guint ceiling;
  guint next = 0;
  guint down_samples = 0;
  guint required_down_samples = WEBRTCPUSH_REMB_DOWN_CONFIRMATIONS;
  gboolean severe_down = FALSE;
  gboolean can_step_down = FALSE;
  gboolean have_pacing;
  gboolean first_frame_sent;
  gboolean recent_idr = FALSE;
  gboolean remb_sane_floor = FALSE;
  gboolean log_remb_sane_floor = FALSE;
  gboolean probe_completed_now = FALSE;
  guint pacing_packets = 0;
  guint pacing_bytes = 0;
  guint pacing_delay_ms = 0;
  gint64 now_us;

  if (!client || bitrate == 0)
    return;
  now_us = g_get_monotonic_time();

  have_pacing = pacing_queue_snapshot(client, track, &pacing_packets,
                                      &pacing_bytes, &pacing_delay_ms);

  g_mutex_lock(&client->lock);
  current = client->target_bitrate;
  first_frame_sent = client->first_frame_sent;
  recent_idr = client->last_idr_sent_us > 0 &&
      now_us - client->last_idr_sent_us <
          (gint64)WEBRTCPUSH_REMB_IDR_GRACE_MS * 1000;
  client->last_remb_bitrate = bitrate;
  can_step_down = now_us - client->last_bitrate_ramp_us >=
      (gint64)WEBRTCPUSH_BITRATE_RAMP_DOWN_MS * 1000;
  /*
   * REMB 在低发送量时容易形成“低码率 -> 低估计 -> 永远无法探测”的闭环。
   * 初次连接主动爬到健康探测目标并保持一段时间，让接收端用真实流量重新估计。
   * 探测完成后仍保留“发送端受限”保护：pacing 健康时，低 REMB 只能阻止
   * 继续上升，不能把发送量越压越低。只有持续积压且不在 IDR 窗口才允许下降。
   */
  if (!client->initial_probe_complete) {
    if (current + WEBRTCPUSH_REMB_DOWN_MIN_STEP >=
        WEBRTCPUSH_HEALTHY_PROBE_TARGET_BPS) {
      if (client->probe_target_reached_us == 0)
        client->probe_target_reached_us = now_us;
      else if (now_us - client->probe_target_reached_us >=
                   (gint64)WEBRTCPUSH_INITIAL_PROBE_HOLD_MS * 1000) {
        client->initial_probe_complete = TRUE;
        probe_completed_now = TRUE;
      }
    }
  }
  estimate_bitrate = bitrate;
  if (first_frame_sent && have_pacing &&
      bitrate < WEBRTCPUSH_REMB_SANE_RAW_MAX_BPS &&
      (pacing_delay_ms <= WEBRTCPUSH_REMB_SANE_PACING_MAX_MS || recent_idr)) {
    estimate_bitrate = WEBRTCPUSH_REMB_SANE_FLOOR_BPS;
    remb_sane_floor = TRUE;
    if (client->last_remb_sane_log_us == 0 ||
        now_us - client->last_remb_sane_log_us >= 5000000) {
      client->last_remb_sane_log_us = now_us;
      log_remb_sane_floor = TRUE;
    }
  }
  usable = clamp_bitrate(
      (guint)(((guint64)estimate_bitrate * WEBRTCPUSH_REMB_UTIL_PERCENT) / 100U));

  if (client->remb_filtered == 0) {
    filtered = estimate_bitrate;
  } else if (estimate_bitrate < client->remb_filtered) {
    /* React to a real fall, but do not mirror every 200ms estimator sample. */
    filtered = (guint)(((guint64)client->remb_filtered * 2U + estimate_bitrate) / 3U);
  } else {
    /* Rising bandwidth: 3:1 权重，跌落后能较快爬回（原 7:1 太慢） */
    filtered = (guint)(((guint64)client->remb_filtered * 3U + estimate_bitrate) / 4U);
  }

  /*
   * 浏览器/手机端的 REMB 有时会在 pacing 几乎为空时给出 500~900kbps
   * 的低估值。这个场景不是网络堵塞，若继续拿旧 filtered 参与 ceiling，
   * 会把 720p 长时间压回 800kbps。保护命中时把 filtered 也抬到 sane floor，
   * 让发送端按“可疑但不堵”的网络慢慢爬升，而不是突然降清晰度。
   */
  if (remb_sane_floor && filtered < estimate_bitrate)
    filtered = estimate_bitrate;

  severe_down = ((guint64)usable * 100U) <
                ((guint64)current * WEBRTCPUSH_REMB_SEVERE_PERCENT);
  if (severe_down) {
    /* Reset the filter too, otherwise its old high value immediately rebounds. */
    filtered = estimate_bitrate;
    ceiling = usable;
    required_down_samples = WEBRTCPUSH_REMB_SEVERE_CONFIRMATIONS;
  } else {
    ceiling = clamp_bitrate(
        (guint)(((guint64)filtered * WEBRTCPUSH_REMB_UTIL_PERCENT) / 100U));
  }
  client->remb_filtered = filtered;
  client->remb_ceiling = ceiling;

  if (ceiling + WEBRTCPUSH_REMB_DOWN_MIN_STEP <= current) {
    client->remb_down_samples++;
    down_samples = client->remb_down_samples;
    if (client->remb_down_samples >= required_down_samples) {
      if (!can_step_down) {
        g_mutex_unlock(&client->lock);
        my_zlog_debug("libdatachannel: REMB down held %u/%u raw=%u filtered=%u ceiling=%u current=%u kbps",
                      down_samples, required_down_samples,
                      bitrate / 1000, filtered / 1000,
                      ceiling / 1000, current / 1000);
        return;
      }
      client->remb_down_samples = 0;
      /* target_bitrate is committed by apply_bitrate() after optional flush. */
      client->last_bitrate_ramp_us = now_us;
      next = bitrate_step_down_toward(current, ceiling);
    }
  } else if (ceiling >= current) {
    client->remb_down_samples = 0;
  }
  g_mutex_unlock(&client->lock);

  if (probe_completed_now) {
    my_zlog_info("libdatachannel: initial bandwidth probe complete at %u kbps; steady sender-limited REMB guard enabled",
                 current / 1000);
  }

  if (log_remb_sane_floor) {
    my_zlog_info("libdatachannel: RTCP REMB sender-limited raw=%u kbps, use=%u kbps (pacing=%u pkt/%u bytes/%ums idr_grace=%d)",
                 bitrate / 1000, estimate_bitrate / 1000,
                 pacing_packets, pacing_bytes, pacing_delay_ms,
                 recent_idr ? 1 : 0);
  }

  if (next) {
    apply_bitrate(client, track, next);
    if (severe_down) {
      my_zlog_info("libdatachannel: RTCP REMB raw=%u%s filtered=%u kbps, encoder=%u kbps ceiling=%u kbps, severe smoothed",
                   bitrate / 1000, remb_sane_floor ? " sane-floor" : "",
                   filtered / 1000, next / 1000, ceiling / 1000);
    } else {
      my_zlog_info("libdatachannel: RTCP REMB raw=%u%s filtered=%u kbps, encoder=%u kbps ceiling=%u kbps after %u samples",
                   bitrate / 1000, remb_sane_floor ? " sane-floor" : "",
                   filtered / 1000, next / 1000, ceiling / 1000,
                   down_samples);
    }
  } else if (ceiling > current) {
    my_zlog_debug("libdatachannel: REMB raw=%u%s filtered=%u ceiling=%u kbps (slow ramp)",
                  bitrate / 1000, remb_sane_floor ? " sane-floor" : "",
                  filtered / 1000, ceiling / 1000);
  } else if (down_samples > 0) {
    my_zlog_debug("libdatachannel: REMB down candidate %u/%u raw=%u%s filtered=%u kbps",
                  down_samples, required_down_samples,
                  bitrate / 1000, remb_sane_floor ? " sane-floor" : "",
                  filtered / 1000);
  }
}

static void maybe_ramp_bitrate(RtcClient *client, int track, gint64 now_us)
{
  guint cur;
  guint tgt;

  if (!client || track < 0)
    return;

  g_mutex_lock(&client->lock);
  cur = client->target_bitrate;
  tgt = client->remb_ceiling;
  if (tgt <= cur ||
      now_us - client->last_bitrate_ramp_us <
          (gint64)WEBRTCPUSH_BITRATE_RAMP_UP_MS * 1000) {
    g_mutex_unlock(&client->lock);
    return;
  }
  client->last_bitrate_ramp_us = now_us;
  g_mutex_unlock(&client->lock);

  ramp_bitrate_toward_ceiling(client, track);
}

static guint32 pts_delta_to_rtp(guint64 delta_ns)
{
  const guint64 second = G_GUINT64_CONSTANT(1000000000);
  guint64 ticks = (delta_ns / second) * 90000;
  ticks += ((delta_ns % second) * 90000) / second;
  return (guint32)ticks;
}

static gpointer pull_thread_main(gpointer data)
{
  RtcClient *client = data;
  const gint pull_ms = (1000 / WEBRTCPUSH_H264_FPS) + 100;
  guint pull_fail_streak = 0;
  gint64 last_pull_warn_us = 0;

  for (;;) {
    gboolean stopping;
    gboolean track_open;
    gboolean need_idr_now;
    const guint8 *frame_data;
    size_t frame_size;
    gboolean is_idr;
    guint64 source_pts;
    RtcH264Frame *queued;

    g_mutex_lock(&client->lock);
    stopping = client->stopping;
    track_open = client->track_open && client->track >= 0;
    need_idr_now = track_open && client->need_idr;
    g_mutex_unlock(&client->lock);
    if (stopping)
      break;
    if (!track_open) {
      g_usleep(20000);
      continue;
    }

    if (need_idr_now)
      mpp_h264_source_request_keyframe(client->h264_source);

    {
      if (!mpp_h264_source_pull(client->h264_source, FALSE, &frame_data,
                                &frame_size, &is_idr, &source_pts,
                                pull_ms)) {
        gint64 now_us = g_get_monotonic_time();
        pull_fail_streak++;
        if (now_us - last_pull_warn_us >= 3000000 ||
            pull_fail_streak == 1 || pull_fail_streak % 20 == 0) {
          my_zlog_warn("libdatachannel: live H264 frame pull timeout (streak=%u)",
                       pull_fail_streak);
          last_pull_warn_us = now_us;
        }
        if (pull_fail_streak >= 20) {
          if (mpp_h264_source_recover(client->h264_source))
            pull_fail_streak = 0;
          g_usleep(200000);
        } else {
          g_usleep(5000);
        }
        continue;
      }
      pull_fail_streak = 0;
    }

    if (!track_open)
      continue;

    if (is_idr) {
      g_mutex_lock(&client->lock);
      if (client->need_idr) {
        client->need_idr = FALSE;
        client->last_idr_sent_us = g_get_monotonic_time();
      }
      g_mutex_unlock(&client->lock);
    }

    queued = rtc_h264_frame_new(frame_data, frame_size, is_idr, source_pts);
    frame_queue_push(client, queued);
  }
  return NULL;
}

static gpointer send_thread_main(gpointer data)
{
  RtcClient *client = data;
  const guint32 timestamp_step = 90000 / WEBRTCPUSH_H264_FPS;
  const gint pop_ms = (1000 / WEBRTCPUSH_H264_FPS) + 200;
  guint send_fail_streak = 0;
  guint64 sent_window_bytes = 0;
  guint sent_window_frames = 0;
  gint64 sent_window_start_us = g_get_monotonic_time();
  gint64 last_pacing_check_us = 0;
  gint64 last_pacing_flush_us = 0;

  for (;;) {
    gboolean stopping;
    int track;
    guint target_bps;
    guint raw_remb_bps;
    guint filtered_remb_bps;
    guint remb_bps;
    guint queue_dropped;
    guint pacing_packets = 0;
    guint pacing_bytes = 0;
    guint pacing_delay_ms = 0;
    guint actual_kbps = 0;
    guint actual_fps_x10 = 0;
    guint32 timestamp;
    RtcH264Frame *frame;
    gint64 now;

    g_mutex_lock(&client->lock);
    stopping = client->stopping;
    g_mutex_unlock(&client->lock);
    if (stopping)
      break;

    g_mutex_lock(&client->lock);
    while (!client->stopping && (!client->track_open || client->track < 0))
      g_cond_wait(&client->track_cond, &client->lock);
    stopping = client->stopping;
    track = client->track;
    g_mutex_unlock(&client->lock);
    if (stopping)
      break;
    if (track < 0)
      continue;

    now = g_get_monotonic_time();
    maybe_ramp_bitrate(client, track, now);

    if (now - last_pacing_check_us >= 100000) {
      last_pacing_check_us = now;
      if (pacing_queue_snapshot(client, track, &pacing_packets,
                                &pacing_bytes, &pacing_delay_ms) &&
          pacing_delay_ms > WEBRTCPUSH_PACING_MAX_QUEUE_MS &&
          client->first_frame_sent &&
          now - last_pacing_flush_us >=
              (gint64)WEBRTCPUSH_PACING_GUARD_COOLDOWN_MS * 1000) {
        guint dropped_frames = 0;
        int dropped;
        gboolean in_idr_grace;
        gboolean should_request_idr;
        gint64 last_idr_us;
        gint64 last_idr_request_us;
        gboolean need_idr_pending;

        g_mutex_lock(&client->lock);
        last_idr_us = client->last_idr_sent_us;
        last_idr_request_us = client->last_idr_request_us;
        need_idr_pending = client->need_idr;
        in_idr_grace =
            last_idr_us > 0 &&
            now - last_idr_us <
                (gint64)WEBRTCPUSH_PACING_IDR_GRACE_MS * 1000;
        g_mutex_unlock(&client->lock);

        if (in_idr_grace) {
          my_zlog_warn("libdatachannel: pacing backlog %u packets/%u bytes ~= %ums; IDR grace, hold queue/bitrate",
                       pacing_packets, pacing_bytes, pacing_delay_ms);
          continue;
        }

        last_pacing_flush_us = now;
        dropped = clear_stale_pacing(client, track, &dropped_frames);
        should_request_idr =
            !need_idr_pending && dropped > 0 &&
            (last_idr_us == 0 ||
             now - last_idr_us >=
                 (gint64)WEBRTCPUSH_PACING_RESYNC_IDR_MS * 1000) &&
            (last_idr_request_us == 0 ||
             now - last_idr_request_us >=
                 (gint64)WEBRTCPUSH_PACING_RESYNC_IDR_MS * 1000);
        if (should_request_idr) {
          request_idr(client, TRUE);
          my_zlog_warn("libdatachannel: pacing backlog %u packets/%u bytes ~= %ums; dropped=%d RTP/%u encoded and request IDR",
                       pacing_packets, pacing_bytes, pacing_delay_ms,
                       dropped > 0 ? dropped : 0, dropped_frames);
        } else {
          my_zlog_warn("libdatachannel: pacing backlog %u packets/%u bytes ~= %ums; dropped=%d RTP/%u encoded without IDR (recent/pending/no-drop)",
                       pacing_packets, pacing_bytes, pacing_delay_ms,
                       dropped > 0 ? dropped : 0, dropped_frames);
        }
      }
    }

    g_mutex_lock(&client->lock);
    if (client->last_stats_log_us == 0 ||
        now - client->last_stats_log_us >=
            (gint64)WEBRTCPUSH_STATS_LOG_INTERVAL_MS * 1000) {
      target_bps = client->target_bitrate;
      raw_remb_bps = client->last_remb_bitrate;
      filtered_remb_bps = client->remb_filtered;
      remb_bps = client->remb_ceiling;
      client->last_stats_log_us = now;
      g_mutex_unlock(&client->lock);
      queue_dropped = client->frame_queue.dropped_total;
      pacing_queue_snapshot(client, track, &pacing_packets,
                            &pacing_bytes, &pacing_delay_ms);
      if (now > sent_window_start_us) {
        guint64 elapsed_us = (guint64)(now - sent_window_start_us);
        actual_kbps = (guint)((sent_window_bytes * 8U * 1000U) / elapsed_us);
        actual_fps_x10 = (guint)(((guint64)sent_window_frames * 10U * 1000000U) /
                                elapsed_us);
      }
      my_zlog_info("libdatachannel: stats encoder=%u actual_h264=%u kbps sent_fps=%u.%u raw_remb=%u filtered_remb=%u ceiling=%u kbps pacing=%u pkt/%u bytes/%ums queue_dropped=%u probe=%s",
                   target_bps / 1000, actual_kbps,
                   actual_fps_x10 / 10, actual_fps_x10 % 10,
                   raw_remb_bps / 1000, filtered_remb_bps / 1000,
                   remb_bps / 1000,
                   pacing_packets, pacing_bytes, pacing_delay_ms,
                   queue_dropped,
                   client->initial_probe_complete ? "done" : "active");
      send_datachannel_mbps(client, actual_kbps);
      /* 修复: MPP CBR失控时actual>>target, pacing必须跟actual否则永远堆积 */
      if (actual_kbps > 0) {
        guint actual_bps = actual_kbps * 1000U;
        guint new_pacing = (guint)((guint64)actual_bps *
                                    WEBRTCPUSH_PACING_HEADROOM_PERCENT / 100U);
        if (new_pacing > WEBRTCPUSH_PACING_MAX_BITRATE)
          new_pacing = WEBRTCPUSH_PACING_MAX_BITRATE;
        g_mutex_lock(&client->send_lock);
        if (client->track == track)
          rtcSetPacingBitrate(track, new_pacing);
        g_mutex_unlock(&client->send_lock);
      }
      sent_window_bytes = 0;
      sent_window_frames = 0;
      sent_window_start_us = now;
    } else {
      g_mutex_unlock(&client->lock);
    }

    frame = frame_queue_pop(client, pop_ms);
    if (!frame) {
      if (client->frame_queue.stopping)
        break;
      continue;
    }

    g_mutex_lock(&client->lock);
    if (frame->pts_ns != G_MAXUINT64) {
      if (!client->source_pts_valid || frame->pts_ns < client->last_source_pts) {
        client->source_pts_base = frame->pts_ns;
        client->last_source_pts = frame->pts_ns;
        client->rtp_timestamp_base = client->rtp_timestamp;
        client->source_pts_valid = TRUE;
      }
      timestamp = client->rtp_timestamp_base +
                  pts_delta_to_rtp(frame->pts_ns - client->source_pts_base);
      client->last_source_pts = frame->pts_ns;
    } else {
      timestamp = client->rtp_timestamp;
    }
    g_mutex_unlock(&client->lock);

    g_mutex_lock(&client->send_lock);
    if (!rtcIsOpen(track) ||
        rtcSetTrackRtpTimestamp(track, timestamp) < 0 ||
        rtcSendMessage(track, (const char *)frame->data, (int)frame->size) < 0) {
      g_mutex_unlock(&client->send_lock);
      send_fail_streak++;
      my_zlog_warn("libdatachannel: rtcSendMessage failed (%u)",
                   send_fail_streak);
      if (send_fail_streak >= WEBRTCPUSH_SEND_FAIL_RESET) {
        my_zlog_error("libdatachannel: send failed %u times, waiting for track reopen",
                      send_fail_streak);
        g_mutex_lock(&client->lock);
        client->track_open = FALSE;
  client->first_frame_sent = FALSE;
        client->need_idr = TRUE;
        g_mutex_unlock(&client->lock);
        signal_track_ready(client);
        send_fail_streak = 0;
      }
      rtc_h264_frame_free(frame);
      g_usleep(send_fail_streak >= 10 ? 200000 : 20000U * send_fail_streak);
      continue;
    }
    g_mutex_unlock(&client->send_lock);
    send_fail_streak = 0;
    if (!client->first_frame_sent) {
      client->first_frame_sent = TRUE;
      my_zlog_info("libdatachannel: first frame sent (%zu bytes), pacing flush armed", frame->size);
    }
    sent_window_bytes += frame->size;
    sent_window_frames++;

    g_mutex_lock(&client->lock);
    client->rtp_timestamp = timestamp + timestamp_step;
    if (frame->is_idr) {
      client->need_idr = FALSE;
      client->last_idr_sent_us = g_get_monotonic_time();
    }
    g_mutex_unlock(&client->lock);

    rtc_h264_frame_free(frame);
  }
  return NULL;
}

static gpointer audio_send_thread_main(gpointer data)
{
  RtcClient *client = data;
  /* Opus 20ms 帧 → 50fps；RTP 时间戳步进 = 48000/50 = 960 */
  const guint32 timestamp_step = WEBRTCPUSH_OPUS_CLOCKRATE / 50;
  const gint pull_ms = 200;
  guint send_fail_streak = 0;

  for (;;) {
    gboolean stopping;
    gboolean audio_send_enabled;
    int track;
    const uint8_t *frame_data;
    size_t frame_size;
    guint64 pts_ns;
    guint32 timestamp;

    g_mutex_lock(&client->lock);
    stopping = client->stopping;
    g_mutex_unlock(&client->lock);
    if (stopping)
      break;

    g_mutex_lock(&client->lock);
    while (!client->stopping &&
           (!client->audio_track_open || client->audio_track < 0 ||
            !client->audio_send_enabled))
      g_cond_wait(&client->track_cond, &client->lock);
    stopping = client->stopping;
    track = client->audio_track;
    audio_send_enabled = client->audio_send_enabled;
    g_mutex_unlock(&client->lock);
    if (stopping)
      break;
    if (track < 0 || !audio_send_enabled)
      continue;

    if (!audio_source_pull(client->audio_source, &frame_data, &frame_size,
                           &pts_ns, pull_ms))
      continue;

    g_mutex_lock(&client->lock);
    if (!client->audio_pts_valid || pts_ns < client->audio_last_pts) {
      client->audio_pts_base = pts_ns;
      client->audio_last_pts = pts_ns;
      client->audio_pts_valid = TRUE;
    }
    timestamp = client->audio_rtp_timestamp +
        (guint32)((pts_ns - client->audio_pts_base) *
                  (guint64)WEBRTCPUSH_OPUS_CLOCKRATE / G_GUINT64_CONSTANT(1000000000));
    client->audio_last_pts = pts_ns;
    g_mutex_unlock(&client->lock);

    g_mutex_lock(&client->send_lock);
    if (!rtcIsOpen(track) ||
        rtcSetTrackRtpTimestamp(track, timestamp) < 0 ||
        rtcSendMessage(track, (const char *)frame_data, (int)frame_size) < 0) {
      g_mutex_unlock(&client->send_lock);
      send_fail_streak++;
      if (send_fail_streak >= WEBRTCPUSH_SEND_FAIL_RESET) {
        g_mutex_lock(&client->lock);
        client->audio_track_open = FALSE;
        g_mutex_unlock(&client->lock);
        send_fail_streak = 0;
      }
      g_usleep(20000);
      continue;
    }
    g_mutex_unlock(&client->send_lock);
    send_fail_streak = 0;

    g_mutex_lock(&client->lock);
    client->audio_rtp_timestamp = timestamp + timestamp_step;
    g_mutex_unlock(&client->lock);
  }
  return NULL;
}

static void offer_video_clear(OfferVideo *video)
{
  g_free(video->mid);
  g_free(video->ice_ufrag);
  memset(video, 0, sizeof(*video));
  video->payload_type = -1;
  video->mline_index = -1;
}

static void offer_audio_clear(OfferAudio *audio)
{
  audio->direction = 0;
  g_free(audio->mid);
  memset(audio, 0, sizeof(*audio));
  audio->payload_type = -1;
}

static gboolean parse_offer(const char *sdp, OfferVideo *video, OfferAudio *audio)
{
  gchar **lines;
  gboolean in_video = FALSE;
  gboolean in_audio = FALSE;
  gint mline = -1;
  gint i;

  memset(video, 0, sizeof(*video));
  video->payload_type = -1;
  video->mline_index = -1;
  memset(audio, 0, sizeof(*audio));
  audio->payload_type = -1;
  audio->found = FALSE;

  lines = g_strsplit(sdp, "\n", -1);
  for (i = 0; lines[i]; i++) {
    gchar *line = g_strchomp(lines[i]);
    if (g_str_has_prefix(line, "m=")) {
      mline++;
      in_video = g_str_has_prefix(line, "m=video ");
      in_audio = g_str_has_prefix(line, "m=audio ");
      if (in_video)
        video->mline_index = mline;
      if (in_audio)
        audio->found = TRUE;
      continue;
    }
    if (!video->ice_ufrag && g_str_has_prefix(line, "a=ice-ufrag:"))
      video->ice_ufrag = g_strdup(line + strlen("a=ice-ufrag:"));
    if (in_video) {
      if (g_str_has_prefix(line, "a=mid:")) {
        g_free(video->mid);
        video->mid = g_strdup(line + strlen("a=mid:"));
      } else if (g_str_has_prefix(line, "a=rtpmap:")) {
        gint pt = -1;
        gchar codec[32] = {0};
        if (sscanf(line, "a=rtpmap:%d %31[^/]/90000", &pt, codec) == 2 &&
            !g_ascii_strcasecmp(codec, "H264") && video->payload_type < 0)
          video->payload_type = pt;
      }
    }
    if (in_audio) {
      /* 手机 audio 固定 recvonly（只收不发），设备 answer 固定 sendonly，
       * 不再解析 offer 方向；仅取 mid 和 opus rtpmap。 */
      if (g_str_has_prefix(line, "a=mid:")) {
        g_free(audio->mid);
        audio->mid = g_strdup(line + strlen("a=mid:"));
      } else if (g_str_has_prefix(line, "a=rtpmap:")) {
        gint pt = -1;
        gchar codec[32] = {0};
        if (sscanf(line, "a=rtpmap:%d %31[^/]/48000", &pt, codec) == 2 &&
            !g_ascii_strcasecmp(codec, "opus") && audio->payload_type < 0)
          audio->payload_type = pt;
      }
    }
  }
  g_strfreev(lines);
  if (!video->mid && video->mline_index >= 0)
    video->mid = g_strdup_printf("%d", video->mline_index);
  if (audio->found && !audio->mid)
    audio->mid = g_strdup_printf("%d", video->mline_index + 1);
  return video->payload_type >= 0 && video->mid && video->ice_ufrag;
}

static void clear_local_ice(RtcClient *client)
{
  gchar *json;
  while ((json = g_queue_pop_head(client->pending_local_ice)) != NULL)
    g_free(json);
}

static void destroy_peer(RtcClient *client)
{
  int pc;
  int track;
  int audio_track;
  int data_channel;

  g_mutex_lock(&client->lock);
  pc = client->pc;
  track = client->track;
  audio_track = client->audio_track;
  data_channel = client->data_channel;
  client->pc = -1;
  client->track = -1;
  client->data_channel = -1;
  client->track_open = FALSE;
  client->first_frame_sent = FALSE;
  client->last_idr_sent_us = 0;
  client->audio_track = -1;
  client->audio_payload_type = -1;
  client->audio_track_open = FALSE;
  client->audio_send_enabled = FALSE;
  if (client->dc_audio_buf)
    g_byte_array_set_size(client->dc_audio_buf, 0);
  client->answer_sent = FALSE;
  client->need_idr = TRUE;
  client->source_pts_valid = FALSE;
  clear_local_ice(client);
  g_clear_pointer(&client->remote_ice_ufrag, g_free);
  g_clear_pointer(&client->video_mid, g_free);
  g_mutex_unlock(&client->lock);
  frame_queue_clear(client);
  signal_track_ready(client);

  g_mutex_lock(&client->send_lock);
  if (track >= 0) {
    rtcClose(track);
    rtcDeleteTrack(track);
  }
  if (audio_track >= 0) {
    rtcClose(audio_track);
    rtcDeleteTrack(audio_track);
  }
  if (data_channel >= 0) {
    rtcClose(data_channel);
    rtcDeleteDataChannel(data_channel);
  }
  if (pc >= 0) {
    rtcClosePeerConnection(pc);
    rtcDeletePeerConnection(pc);
  }
  g_mutex_unlock(&client->send_lock);
}

static gboolean setup_track(RtcClient *client, int pc,
                            const OfferVideo *video, int *track_out)
{
  rtcTrackInit track_init;
  rtcPacketizerInit packetizer;
  guint32 ssrc = g_random_int();
  int track;

  if (ssrc == 0)
    ssrc = 1;
  memset(&track_init, 0, sizeof(track_init));
  track_init.direction = RTC_DIRECTION_SENDONLY;
  track_init.codec = RTC_CODEC_H264;
  track_init.payloadType = video->payload_type;
  track_init.ssrc = ssrc;
  track_init.mid = video->mid;
  track_init.name = "car-video";
  track_init.msid = "car-stream";
  track_init.trackId = "car-video";
  {
    gchar *profile = build_h264_profile_fmtp(client);
    track_init.profile = profile;
    track = rtcAddTrackEx(pc, &track_init);
    g_free(profile);
  }
  if (track < 0)
    return FALSE;

  rtcSetUserPointer(track, client);
  rtcSetOpenCallback(track, on_track_open);
  rtcSetClosedCallback(track, on_track_closed);
  rtcSetErrorCallback(track, on_track_error);

  memset(&packetizer, 0, sizeof(packetizer));
  packetizer.ssrc = ssrc;
  packetizer.cname = "car-video";
  packetizer.payloadType = (guint8)video->payload_type;
  packetizer.clockRate = 90000;
  packetizer.sequenceNumber = (guint16)g_random_int();
  packetizer.timestamp = g_random_int();
  packetizer.maxFragmentSize = WEBRTCPUSH_RTP_MAX_FRAGMENT;
  packetizer.nalSeparator = RTC_NAL_SEPARATOR_START_SEQUENCE;

  if (rtcSetH264Packetizer(track, &packetizer) < 0 ||
      rtcChainRtcpSrReporter(track) < 0 ||
      rtcChainRtcpNackResponder(track, WEBRTCPUSH_NACK_PACKETS) < 0 ||
      rtcChainPliHandler(track, on_pli) < 0 ||
      rtcChainRembHandler(track, on_remb) < 0 ||
      rtcChainPacingHandler(track,
                            pacing_bitrate_for_encoder(WEBRTCPUSH_INITIAL_BITRATE),
                            WEBRTCPUSH_PACING_INTERVAL_MS) < 0) {
    rtcDeleteTrack(track);
    return FALSE;
  }

  g_mutex_lock(&client->lock);
  client->rtp_timestamp = packetizer.timestamp;
  client->rtp_timestamp_base = packetizer.timestamp;
  client->source_pts_valid = FALSE;
  g_mutex_unlock(&client->lock);
  *track_out = track;
  return TRUE;
}

static gboolean setup_audio_track(RtcClient *client, int pc,
                                  const OfferAudio *audio, int *track_out)
{
  rtcTrackInit track_init;
  rtcPacketizerInit packetizer;
  guint32 ssrc = g_random_int();
  gboolean send_audio;
  int track;

  if (ssrc == 0)
    ssrc = 1;
  memset(&track_init, 0, sizeof(track_init));
  /* 手机端 audio 为 recvonly（只收不发），设备固定 sendonly 推 Opus 给手机。
   * 手机→设备的音频走 DataChannel 发 MP4，不再通过 RTP 接收。 */
  send_audio = client->audio_source != NULL;
  if (send_audio)
    track_init.direction = RTC_DIRECTION_SENDONLY;
  else
    track_init.direction = RTC_DIRECTION_INACTIVE;

  my_zlog_info("libdatachannel: audio track direction: answer=%d (sendonly)",
               (int)track_init.direction);
  track_init.codec = RTC_CODEC_OPUS;
  track_init.payloadType = audio->payload_type;
  track_init.ssrc = ssrc;
  track_init.mid = audio->mid;
  track_init.name = "car-audio";
  track_init.msid = "car-stream";
  track_init.trackId = "car-audio";
  track = rtcAddTrackEx(pc, &track_init);
  if (track < 0)
    return FALSE;

  rtcSetUserPointer(track, client);
  rtcSetOpenCallback(track, on_track_open);
  rtcSetClosedCallback(track, on_track_closed);
  rtcSetErrorCallback(track, on_track_error);

  memset(&packetizer, 0, sizeof(packetizer));
  packetizer.ssrc = ssrc;
  packetizer.cname = "car-audio";
  packetizer.payloadType = (guint8)audio->payload_type;
  packetizer.clockRate = WEBRTCPUSH_OPUS_CLOCKRATE;
  packetizer.sequenceNumber = (guint16)g_random_int();
  packetizer.timestamp = g_random_int();

  if (send_audio) {
    if (rtcSetOpusPacketizer(track, &packetizer) < 0 ||
        rtcChainRtcpSrReporter(track) < 0 ||
        rtcChainRtcpNackResponder(track, WEBRTCPUSH_NACK_PACKETS) < 0) {
      rtcDeleteTrack(track);
      return FALSE;
    }
  }

  g_mutex_lock(&client->lock);
  client->audio_rtp_timestamp = packetizer.timestamp;
  client->audio_payload_type = audio->payload_type;
  client->audio_pts_valid = FALSE;
  client->audio_send_enabled = send_audio;
  g_mutex_unlock(&client->lock);
  *track_out = track;
  return TRUE;
}

static void flush_remote_ice(RtcClient *client)
{
  for (;;) {
    PendingRemoteIce *ice;
    int pc;
    gboolean ufrag_match;
    g_mutex_lock(&client->lock);
    ice = g_queue_pop_head(client->pending_remote_ice);
    pc = client->pc;
    ufrag_match = !ice || !ice->ufrag || !client->remote_ice_ufrag ||
        g_strcmp0(ice->ufrag, client->remote_ice_ufrag) == 0;
    g_mutex_unlock(&client->lock);
    if (!ice)
      break;
    if (pc >= 0 && ufrag_match) {
      if (rtcAddRemoteCandidate(pc, ice->candidate, ice->mid) < 0)
        my_zlog_warn("libdatachannel: rejected queued remote ICE candidate ufrag=%s",
                     ice->ufrag ? ice->ufrag : "?");
    } else if (!ufrag_match) {
      my_zlog_info("libdatachannel: discard stale remote ICE candidate ufrag=%s current=%s",
                   ice->ufrag ? ice->ufrag : "?",
                   client->remote_ice_ufrag ? client->remote_ice_ufrag : "?");
    }
    free_pending_remote_ice(ice);
  }
}

int rtc_client_start(AppState *app)
{
  RtcClient *client;
  char *error_message = NULL;
  const char *video_device = WEBRTCPUSH_VIDEO_DEVICE;

  if (!app)
    return -1;
  if (app->video_device && app->video_device[0])
    video_device = app->video_device;

  client = g_new0(RtcClient, 1);
  client->app = app;
  client->pc = -1;
  client->track = -1;
  client->audio_track = -1;
  client->audio_payload_type = -1;
  client->data_channel = -1;
  client->target_bitrate = WEBRTCPUSH_INITIAL_BITRATE;
  client->applied_mpp_bitrate = WEBRTCPUSH_INITIAL_BITRATE;
  client->remb_ceiling = WEBRTCPUSH_INITIAL_BITRATE;
  client->initial_probe_complete = FALSE;
  client->probe_target_reached_us = 0;
  client->last_remb_bitrate = 0;
  client->remb_filtered = 0;
  client->remb_down_samples = 0;
  client->last_bitrate_ramp_us = g_get_monotonic_time();
  client->last_mpp_bitrate_apply_us = client->last_bitrate_ramp_us;
  client->pending_local_ice = g_queue_new();
  client->pending_remote_ice = g_queue_new();
  g_mutex_init(&client->lock);
  g_mutex_init(&client->send_lock);
  g_cond_init(&client->track_cond);

  client->h264_source = mpp_h264_source_start(video_device, &error_message);
  if (!client->h264_source) {
    my_zlog_error("libdatachannel: live encoder start failed: %s",
                  error_message ? error_message : "unknown");
    g_free(error_message);
    g_queue_free(client->pending_local_ice);
    g_queue_free(client->pending_remote_ice);
    g_mutex_clear(&client->send_lock);
    g_mutex_clear(&client->lock);
    g_cond_clear(&client->track_cond);
    g_free(client);
    return -1;
  }

  app->rtc_client = client;
  frame_queue_init(&client->frame_queue, WEBRTCPUSH_SEND_QUEUE_DEPTH);
  client->pull_thread = g_thread_new("rtc-h264-pull", pull_thread_main, client);
  client->send_thread = g_thread_new("rtc-h264-send", send_thread_main, client);
  mpp_h264_source_set_bitrate(client->h264_source, client->target_bitrate);

  /* 音频源（可选；app->enable_audio 由 setup_alsa_device 探测决定） */
  if (app->enable_audio && app->audio_alsa_device && app->audio_alsa_device[0]) {
    char *audio_err = NULL;
    client->audio_source = audio_source_start(app->audio_alsa_device, &audio_err);
    if (client->audio_source) {
      client->audio_send_thread = g_thread_new("rtc-opus-send", audio_send_thread_main, client);
      my_zlog_info("libdatachannel: audio source started device=%s", app->audio_alsa_device);
    } else {
      my_zlog_warn("libdatachannel: audio source start failed: %s", audio_err ? audio_err : "?");
      g_free(audio_err);
    }
  }

  /* 音频播放（手机->设备） */
  {
    char *play_err = NULL;
    client->audio_sink = audio_sink_start(WEBRTCPUSH_AUDIO_PLAYBACK_DEVICE, &play_err);
    if (client->audio_sink) {
      my_zlog_info("libdatachannel: audio sink started device=%s", WEBRTCPUSH_AUDIO_PLAYBACK_DEVICE);
      if (app->dev_room && app->dev_room[0]) {
        client->volume_ctrl = volume_control_start(app->dev_room, client->audio_sink);
        my_zlog_info("libdatachannel: volume control started vehicle=%s", app->dev_room);
      }
    } else {
      my_zlog_warn("libdatachannel: audio sink start failed: %s", play_err ? play_err : "?");
      g_free(play_err);
    }
  }
  my_zlog_info("libdatachannel: MPP live source started device=%s mpp=%d fps=%d threads=pull+send queue=%u",
               video_device,
               mpp_h264_source_is_mpp_encoder(client->h264_source) ? 1 : 0,
               WEBRTCPUSH_H264_FPS,
               WEBRTCPUSH_SEND_QUEUE_DEPTH);
  return 0;
}

void rtc_client_reset(AppState *app)
{
  RtcClient *client = client_from_app(app);
  if (client) {
    destroy_peer(client);
    g_mutex_lock(&client->lock);
    g_queue_clear_full(client->pending_remote_ice, free_pending_remote_ice);
    g_mutex_unlock(&client->lock);
  }
}

void rtc_client_stop(AppState *app)
{
  RtcClient *client = client_from_app(app);
  if (!client)
    return;

  destroy_peer(client);
  g_mutex_lock(&client->lock);
  client->stopping = TRUE;
  g_cond_broadcast(&client->track_cond);
  g_mutex_unlock(&client->lock);
  frame_queue_stop(&client->frame_queue);
  if (client->pull_thread)
    g_thread_join(client->pull_thread);
  if (client->send_thread)
    g_thread_join(client->send_thread);
  if (client->audio_send_thread)
    g_thread_join(client->audio_send_thread);
  if (client->volume_ctrl)
    volume_control_stop(client->volume_ctrl);
  if (client->audio_sink)
    audio_sink_stop(client->audio_sink);
  if (client->audio_source)
    audio_source_stop(client->audio_source);

  g_queue_free_full(client->pending_remote_ice, free_pending_remote_ice);
  g_mutex_lock(&client->lock);
  clear_local_ice(client);
  g_mutex_unlock(&client->lock);
  g_queue_free(client->pending_local_ice);
  if (client->dc_audio_buf)
    g_byte_array_unref(client->dc_audio_buf);
  /* 停止播放队列线程 */
  if (client->dc_play_queue) {
    client->dc_play_quit = TRUE;
    g_async_queue_push(client->dc_play_queue, GINT_TO_POINTER(-1)); /* 唤醒 */
    if (client->dc_play_thread)
      g_thread_join(client->dc_play_thread);
    /* 清空队列剩余任务 */
    while (1) {
      gpointer p = g_async_queue_try_pop(client->dc_play_queue);
      if (!p || p == GINT_TO_POINTER(-1)) break;
      DcMp3PlayTask *t = p;
      if (t) { g_free(t->data); g_free(t); }
    }
    g_async_queue_unref(client->dc_play_queue);
    client->dc_play_queue = NULL;
    client->dc_play_thread = NULL;
  }
  mpp_h264_source_stop(client->h264_source);
  frame_queue_fini(&client->frame_queue);
  g_cond_clear(&client->track_cond);
  g_mutex_clear(&client->send_lock);
  g_mutex_clear(&client->lock);
  app->rtc_client = NULL;
  g_free(client);
}

int rtc_client_handle_offer(AppState *app, const char *sdp)
{
  RtcClient *client = client_from_app(app);
  OfferVideo video;
  OfferAudio audio;
  rtcConfiguration config;
  int pc;
  int track = -1;
  int audio_track = -1;
  int data_channel = -1;
  gboolean duplicate;

  if (!client || !sdp)
    return -1;
  if (!parse_offer(sdp, &video, &audio)) {
    my_zlog_error("libdatachannel: phone offer has no usable H264 video");
    my_zlog_error("libdatachannel: SDP dump BEGIN:\n%s", sdp);
    offer_video_clear(&video);
    offer_audio_clear(&audio);
    return -1;
  }

  g_mutex_lock(&client->lock);
  duplicate = client->pc >= 0 &&
      g_strcmp0(client->remote_ice_ufrag, video.ice_ufrag) == 0;
  g_mutex_unlock(&client->lock);
  if (duplicate) {
    offer_video_clear(&video);
    offer_audio_clear(&audio);
    return 0;
  }

  destroy_peer(client);
  memset(&config, 0, sizeof(config));
  config.iceServers = s_ice_servers;
  config.iceServersCount = (int)G_N_ELEMENTS(s_ice_servers);
  config.certificateType = RTC_CERTIFICATE_ECDSA;
  config.forceMediaTransport = TRUE; /* 预分配 SRTP 媒体通道；DataChannel 使用同一个 PC 的 SCTP m-line */
  config.mtu = 1200;
  pc = rtcCreatePeerConnection(&config);
  if (pc < 0) {
    offer_video_clear(&video);
    return -1;
  }

  rtcSetUserPointer(pc, client);
  rtcSetLocalDescriptionCallback(pc, on_local_description);
  rtcSetLocalCandidateCallback(pc, on_local_candidate);
  rtcSetStateChangeCallback(pc, on_state_change);
  rtcSetDataChannelCallback(pc, on_incoming_data_channel);

#if WEBRTCPUSH_ENABLE_DATACHANNEL
  data_channel = rtcCreateDataChannel(pc, "myDataChannel");
  if (data_channel >= 0) {
    setup_data_channel_callbacks(client, data_channel);
    my_zlog_info("libdatachannel: DataChannel created label=myDataChannel");
  } else {
    my_zlog_warn("libdatachannel: failed to create DataChannel");
  }
#endif

  if (!setup_track(client, pc, &video, &track)) {
    if (data_channel >= 0)
      rtcDeleteDataChannel(data_channel);
    rtcDeletePeerConnection(pc);
    offer_video_clear(&video);
    offer_audio_clear(&audio);
    my_zlog_error("libdatachannel: failed to create H264 track/handlers");
    return -1;
  }

  /* 若手机 offer 包含 audio m-line，按本机麦克风/喇叭能力协商方向。 */
  if (audio.found && audio.payload_type >= 0 &&
      (client->audio_source || client->audio_sink)) {
    if (setup_audio_track(client, pc, &audio, &audio_track)) {
      my_zlog_info("libdatachannel: Opus audio track created mid=%s pt=%d dir=%d",
                   audio.mid, audio.payload_type, audio.direction);
    } else {
      my_zlog_warn("libdatachannel: failed to create Opus audio track");
    }
  }


  g_mutex_lock(&client->lock);
  client->pc = pc;
  client->track = track;
  client->audio_track = audio_track;
  client->data_channel = data_channel;
  client->video_mline_index = video.mline_index;
  client->video_mid = g_strdup(video.mid);
  client->remote_ice_ufrag = g_strdup(video.ice_ufrag);
  client->target_bitrate = WEBRTCPUSH_INITIAL_BITRATE;
  client->applied_mpp_bitrate = WEBRTCPUSH_INITIAL_BITRATE;
  client->remb_ceiling = WEBRTCPUSH_INITIAL_BITRATE;
  client->initial_probe_complete = FALSE;
  client->probe_target_reached_us = 0;
  client->last_remb_bitrate = 0;
  client->remb_filtered = 0;
  client->remb_down_samples = 0;
  client->last_bitrate_ramp_us = g_get_monotonic_time();
  client->last_mpp_bitrate_apply_us = client->last_bitrate_ramp_us;
  client->need_idr = TRUE;          /* 内联，避免 request_idr 二次加锁死锁 */
  client->last_idr_request_us = g_get_monotonic_time();
  client->source_pts_valid = FALSE;
  g_mutex_unlock(&client->lock);
  mpp_h264_source_set_bitrate(client->h264_source,
                              WEBRTCPUSH_INITIAL_BITRATE);

  my_zlog_info("libdatachannel: applying offer mid=%s pt=%d ufrag=%s",
               video.mid, video.payload_type, video.ice_ufrag);
  if (rtcSetRemoteDescription(pc, sdp, "offer") < 0) {
    my_zlog_error("libdatachannel: rtcSetRemoteDescription failed");
    offer_video_clear(&video);
    offer_audio_clear(&audio);
    destroy_peer(client);
    return -1;
  }
  flush_remote_ice(client);
  offer_video_clear(&video);
  offer_audio_clear(&audio);
  return 0;
}

void rtc_client_handle_remote_ice(AppState *app, JsonObject *candidate_obj)
{
  RtcClient *client = client_from_app(app);
  const gchar *candidate;
  const gchar *mid = NULL;
  const gchar *ufrag = NULL;
  gchar *candidate_ufrag = NULL;
  gboolean queued = FALSE;
  int pc;

  if (!client || !candidate_obj ||
      !json_object_has_member(candidate_obj, "candidate"))
    return;
  candidate = json_object_get_string_member(candidate_obj, "candidate");
  if (!candidate || !candidate[0])
    return;
  if (json_object_has_member(candidate_obj, "sdpMid")) {
    JsonNode *node = json_object_get_member(candidate_obj, "sdpMid");
    if (JSON_NODE_HOLDS_VALUE(node) &&
        json_node_get_value_type(node) == G_TYPE_STRING)
      mid = json_object_get_string_member(candidate_obj, "sdpMid");
  }
  if (json_object_has_member(candidate_obj, "usernameFragment")) {
    JsonNode *node = json_object_get_member(candidate_obj, "usernameFragment");
    if (JSON_NODE_HOLDS_VALUE(node) &&
        json_node_get_value_type(node) == G_TYPE_STRING)
      ufrag = json_object_get_string_member(candidate_obj, "usernameFragment");
  } else if (json_object_has_member(candidate_obj, "ufrag")) {
    JsonNode *node = json_object_get_member(candidate_obj, "ufrag");
    if (JSON_NODE_HOLDS_VALUE(node) &&
        json_node_get_value_type(node) == G_TYPE_STRING)
      ufrag = json_object_get_string_member(candidate_obj, "ufrag");
  }
  if (!ufrag) {
    const gchar *tag = strstr(candidate, " ufrag ");
    if (tag) {
      const gchar *end;
      tag += strlen(" ufrag ");
      end = strchr(tag, ' ');
      candidate_ufrag = end ? g_strndup(tag, (gsize)(end - tag)) : g_strdup(tag);
      ufrag = candidate_ufrag;
    }
  }

  g_mutex_lock(&client->lock);
  pc = client->pc;
  if (pc < 0 ||
      (ufrag && client->remote_ice_ufrag &&
       g_strcmp0(ufrag, client->remote_ice_ufrag) != 0)) {
    PendingRemoteIce *pending = g_new0(PendingRemoteIce, 1);
    pending->candidate = g_strdup(candidate);
    pending->mid = g_strdup(mid);
    pending->ufrag = g_strdup(ufrag);
    g_queue_push_tail(client->pending_remote_ice, pending);
    pc = -1;
    queued = TRUE;
  }
  g_mutex_unlock(&client->lock);
  if (pc >= 0 && rtcAddRemoteCandidate(pc, candidate, mid) < 0)
    my_zlog_warn("libdatachannel: rejected remote ICE candidate");
  else if (queued)
    my_zlog_info("libdatachannel: queued remote ICE candidate for ufrag=%s",
                 ufrag ? ufrag : "?");
  g_free(candidate_ufrag);
}

gboolean rtc_client_is_active(AppState *app)
{
  RtcClient *client = client_from_app(app);
  gboolean active = FALSE;
  if (!client)
    return FALSE;
  g_mutex_lock(&client->lock);
  active = client->pc >= 0;
  g_mutex_unlock(&client->lock);
  return active;
}

gboolean rtc_client_media_open(AppState *app)
{
  RtcClient *client = client_from_app(app);
  gboolean open = FALSE;
  if (!client)
    return FALSE;
  g_mutex_lock(&client->lock);
  open = client->track_open;
  g_mutex_unlock(&client->lock);
  return open;
}
