#include "rtc_client.h"

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

#include <rtc/rtc.h>

#include <stdio.h>
#include <string.h>

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

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

typedef struct {
  gint payload_type;
  gchar *mid;
  gboolean found;
} 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 {
  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 影响 */
  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 remb_ceiling;
  guint last_remb_bitrate;
  guint remb_filtered;
  guint remb_down_samples;
  gint64 last_bitrate_ramp_us;
  gint64 last_idr_request_us;
  gint64 last_stats_log_us;
  gchar *remote_ice_ufrag;
  gchar *video_mid;
  gint video_mline_index;

  /* 音频 Opus track */
  AudioSource *audio_source;
  int audio_track;
  gboolean audio_track_open;
  GThread *audio_send_thread;
  guint32 audio_rtp_timestamp;
  guint64 audio_pts_base;
  guint64 audio_last_pts;
  gboolean audio_pts_valid;

  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 void frame_queue_clear(RtcClient *client)
{
  RtcFrameQueue *q = &client->frame_queue;
  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);
}

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

  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) {
      q->dropped_total++;
      rtc_h264_frame_free(old);
    }
  }
  if (overflowed && !frame->is_idr) {
    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; request clean IDR");
  }
}

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

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

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;
    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;
    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;
  if (!client || client->track != track)
    return;
  requested = request_idr(client, FALSE);
  if (!requested) {
    my_zlog_debug("libdatachannel: PLI throttled, skip duplicate IDR request");
    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);
  frame_queue_clear(client);
  my_zlog_info("libdatachannel: PLI received, dropped %d queued RTP packet(s), next is IDR",
               dropped > 0 ? dropped : 0);
}

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)
{
  int dropped = 0;

  request_idr(client, TRUE);
  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);
  frame_queue_clear(client);
  return dropped;
}

static void apply_bitrate(RtcClient *client, int track, guint bitrate)
{
  guint b;

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

  g_mutex_lock(&client->lock);
  client->target_bitrate = b;
  g_mutex_unlock(&client->lock);

  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);
  }
  mpp_h264_source_set_bitrate(client->h264_source, b);
}

/* 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 * 12U) / 100U);
  if (step < 50000U)
    step = 50000U;
  if (step > 250000U)
    step = 250000U;
  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 void RTC_API on_remb(int track, unsigned int bitrate, void *ptr)
{
  RtcClient *client = ptr;
  guint current;
  guint usable;
  guint filtered;
  guint ceiling;
  guint next = 0;
  guint down_samples = 0;
  gboolean severe_down = FALSE;
  gint64 now_us;
  int dropped = 0;

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

  usable = clamp_bitrate(
      (guint)(((guint64)bitrate * WEBRTCPUSH_REMB_UTIL_PERCENT) / 100U));

  g_mutex_lock(&client->lock);
  current = client->target_bitrate;
  client->last_remb_bitrate = bitrate;

  if (client->remb_filtered == 0) {
    filtered = bitrate;
  } else if (bitrate < client->remb_filtered) {
    /* React to a real fall, but do not mirror every 200ms estimator sample. */
    filtered = (guint)(((guint64)client->remb_filtered * 2U + bitrate) / 3U);
  } else {
    /* Rising bandwidth needs sustained evidence. */
    filtered = (guint)(((guint64)client->remb_filtered * 7U + bitrate) / 8U);
  }

  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 = bitrate;
    ceiling = usable;
    client->remb_down_samples = 0;
  } 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) {
    if (!severe_down)
      client->remb_down_samples++;
    down_samples = client->remb_down_samples;
    if (severe_down ||
        client->remb_down_samples >= WEBRTCPUSH_REMB_DOWN_CONFIRMATIONS) {
      client->remb_down_samples = 0;
      client->target_bitrate = ceiling;
      client->last_bitrate_ramp_us = now_us;
      next = ceiling;
    }
    if (severe_down) {
      client->need_idr = TRUE;
      client->last_idr_request_us = now_us;
      client->source_pts_valid = FALSE;
    }
  } else if (ceiling >= current) {
    client->remb_down_samples = 0;
  }
  g_mutex_unlock(&client->lock);

  if (next) {
    g_mutex_lock(&client->send_lock);
    if (client->track == track) {
      if (severe_down) {
        dropped = rtcClearPacingQueue(track);
        frame_queue_clear(client);
      }
      rtcSetPacingBitrate(track, pacing_bitrate_for_encoder(next));
    }
    g_mutex_unlock(&client->send_lock);
    mpp_h264_source_set_bitrate(client->h264_source, next);
    if (severe_down) {
      my_zlog_info("libdatachannel: REMB raw=%u filtered=%u kbps, encoder=%u pacing=%u kbps, resync dropped=%d RTP packets",
                   bitrate / 1000, filtered / 1000, next / 1000,
                   pacing_bitrate_for_encoder(next) / 1000,
                   dropped > 0 ? dropped : 0);
    } else {
      my_zlog_info("libdatachannel: REMB raw=%u filtered=%u kbps, encoder=%u pacing=%u kbps after %u samples",
                   bitrate / 1000, filtered / 1000, next / 1000,
                   pacing_bitrate_for_encoder(next) / 1000, down_samples);
    }
  } else if (ceiling > current) {
    my_zlog_debug("libdatachannel: REMB raw=%u filtered=%u ceiling=%u kbps (slow ramp)",
                  bitrate / 1000, filtered / 1000, ceiling / 1000);
  } else if (down_samples > 0) {
    my_zlog_debug("libdatachannel: REMB down candidate %u/%u raw=%u filtered=%u kbps",
                  down_samples, WEBRTCPUSH_REMB_DOWN_CONFIRMATIONS,
                  bitrate / 1000, 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;
  guint idr_discard_streak = 0;
  gint64 last_pull_warn_us = 0;

  for (;;) {
    gboolean stopping;
    gboolean track_open;
    gboolean force_idr;
    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;
    force_idr = track_open && client->need_idr;
    g_mutex_unlock(&client->lock);
    if (stopping)
      break;

    {
      gint pull_timeout = force_idr ? 2500 : pull_ms;

      if (!mpp_h264_source_pull(client->h264_source, force_idr, &frame_data,
                                &frame_size, &is_idr, &source_pts,
                                pull_timeout)) {
        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;

    g_mutex_lock(&client->lock);
    need_idr_now = client->need_idr;
    g_mutex_unlock(&client->lock);

    if (need_idr_now && !is_idr) {
      idr_discard_streak++;
      if (idr_discard_streak >= WEBRTCPUSH_H264_FPS) {
        g_mutex_lock(&client->lock);
        client->need_idr = FALSE;
        g_mutex_unlock(&client->lock);
        idr_discard_streak = 0;
        my_zlog_warn("libdatachannel: IDR wait >2s, resume with latest frames");
      }
      continue;
    }
    idr_discard_streak = 0;

    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) {
        int dropped = clear_stale_pacing(client, track);
        last_pacing_flush_us = now;
        my_zlog_warn("libdatachannel: pacing backlog %u packets/%u bytes ~= %ums; dropped=%d and request IDR",
                     pacing_packets, pacing_bytes, pacing_delay_ms,
                     dropped > 0 ? dropped : 0);
      }
    }

    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",
                   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);
      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;
    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;
    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))
      g_cond_wait(&client->track_cond, &client->lock);
    stopping = client->stopping;
    track = client->audio_track;
    g_mutex_unlock(&client->lock);
    if (stopping)
      break;
    if (track < 0)
      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)
{
  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) {
      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;

  g_mutex_lock(&client->lock);
  pc = client->pc;
  track = client->track;
  audio_track = client->audio_track;
  client->pc = -1;
  client->track = -1;
  client->track_open = FALSE;
  client->first_frame_sent = FALSE;
  client->audio_track = -1;
  client->audio_track_open = FALSE;
  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 (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();
  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_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 (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_pts_valid = FALSE;
  g_mutex_unlock(&client->lock);
  *track_out = track;
  return TRUE;
}

static void flush_remote_ice(RtcClient *client)
{
  for (;;) {
    PendingRemoteIce *ice;
    int pc;
    g_mutex_lock(&client->lock);
    ice = g_queue_pop_head(client->pending_remote_ice);
    pc = client->pc;
    g_mutex_unlock(&client->lock);
    if (!ice)
      break;
    if (pc >= 0)
      rtcAddRemoteCandidate(pc, ice->candidate, ice->mid);
    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->target_bitrate = WEBRTCPUSH_INITIAL_BITRATE;
  client->remb_ceiling = WEBRTCPUSH_INITIAL_BITRATE;
  client->last_remb_bitrate = 0;
  client->remb_filtered = 0;
  client->remb_down_samples = 0;
  client->last_bitrate_ramp_us = g_get_monotonic_time();
  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);
    }
  }
  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);
}

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->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);
  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;
  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 媒体通道，与 ENABLE_DATACHANNEL=0 一致 */
  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);
  if (!setup_track(client, pc, &video, &track)) {
    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 且设备端音频源已启动，建 Opus track */
  if (audio.found && audio.payload_type >= 0 && client->audio_source) {
    if (setup_audio_track(client, pc, &audio, &audio_track)) {
      my_zlog_info("libdatachannel: Opus audio track created mid=%s pt=%d",
                   audio.mid, audio.payload_type);
    } 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->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->remb_ceiling = WEBRTCPUSH_INITIAL_BITRATE;
  client->last_remb_bitrate = 0;
  client->remb_filtered = 0;
  client->remb_down_samples = 0;
  client->last_bitrate_ramp_us = g_get_monotonic_time();
  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;
  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");
  }

  g_mutex_lock(&client->lock);
  pc = client->pc;
  if (pc < 0) {
    PendingRemoteIce *pending = g_new0(PendingRemoteIce, 1);
    pending->candidate = g_strdup(candidate);
    pending->mid = g_strdup(mid);
    g_queue_push_tail(client->pending_remote_ice, pending);
  }
  g_mutex_unlock(&client->lock);
  if (pc >= 0 && rtcAddRemoteCandidate(pc, candidate, mid) < 0)
    my_zlog_warn("libdatachannel: rejected remote ICE candidate");
}

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