#include "rtc_client.h"

#include "mpp_h264_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 {
  AppState *app;
  GMutex lock;
  GMutex send_lock;
  GThread *sender_thread;
  MppH264Source *h264_source;
  gboolean stopping;
  gboolean track_open;
  gboolean need_idr;
  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;
  gint64 last_bitrate_ramp_us;
  gchar *remote_ice_ufrag;
  gchar *video_mid;
  gint video_mline_index;
  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 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 void RTC_API on_local_description(int pc, const char *sdp,
                                         const char *type, void *ptr)
{
  RtcClient *client = ptr;
  gchar *json;
  (void)pc;

  if (!client || !sdp || g_strcmp0(type, "answer") != 0)
    return;
  json = build_description_json(client->app, "answer", sdp);
  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 RTC_API on_state_change(int pc, rtcState state, void *ptr)
{
  RtcClient *client = ptr;
  (void)pc;
  my_zlog_info("libdatachannel: peer state=%s", rtc_state_name(state));
  if (!client)
    return;
  if (state == RTC_CONNECTED) {
    g_mutex_lock(&client->lock);
    client->need_idr = TRUE;
    client->source_pts_valid = FALSE;
    g_mutex_unlock(&client->lock);
  }
}

static void RTC_API on_track_open(int track, void *ptr)
{
  RtcClient *client = ptr;
  if (!client)
    return;
  g_mutex_lock(&client->lock);
  if (client->track == track) {
    client->track_open = TRUE;
    client->need_idr = TRUE;
    client->source_pts_valid = FALSE;
  }
  g_mutex_unlock(&client->lock);
  my_zlog_info("libdatachannel: H264 track open, requesting SPS/PPS+IDR");
}

static void RTC_API on_track_closed(int track, void *ptr)
{
  RtcClient *client = ptr;
  if (!client)
    return;
  g_mutex_lock(&client->lock);
  if (client->track == track)
    client->track_open = FALSE;
  g_mutex_unlock(&client->lock);
  my_zlog_warn("libdatachannel: H264 track closed");
}

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;
  int dropped = 0;
  if (!client)
    return;
  g_mutex_lock(&client->lock);
  client->need_idr = TRUE;
  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);
  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 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);
}

/* 与 gst_webrtc_pipeline::ramp_video_bitrate 同档升降步长 */
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;

  diff = (cur > tgt) ? (cur - tgt) : (tgt - cur);
  if (cur < tgt) {
    step = diff / 8;
    if (step < 50000)
      step = 50000;
    if (step > 150000)
      step = 150000;
    next = cur + step;
    if (next > tgt)
      next = tgt;
  } else {
    step = diff / 3;
    if (step < 100000)
      step = 100000;
    if (step > 600000)
      step = 600000;
    next = (cur > step) ? (cur - step) : 0;
    if (next < tgt)
      next = tgt;
  }

  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 ceiling;
  guint next = 0;
  gboolean severe_down = FALSE;
  int dropped = 0;

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

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

  g_mutex_lock(&client->lock);
  current = client->target_bitrate;
  client->remb_ceiling = ceiling;
  if (ceiling < current) {
    client->target_bitrate = ceiling;
    next = ceiling;
    severe_down = ((guint64)ceiling * 100) < ((guint64)current * 75);
    if (severe_down) {
      client->need_idr = TRUE;
      client->source_pts_valid = FALSE;
    }
  }
  g_mutex_unlock(&client->lock);

  if (next) {
    g_mutex_lock(&client->send_lock);
    if (client->track == track) {
      if (severe_down)
        dropped = rtcClearPacingQueue(track);
      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=%u kbps, encoder=%u kbps pacing=%u kbps, resync dropped=%d RTP packets",
                   bitrate / 1000, next / 1000,
                   pacing_bitrate_for_encoder(next) / 1000,
                   dropped > 0 ? dropped : 0);
    } else {
      my_zlog_info("libdatachannel: REMB=%u kbps, encoder=%u kbps pacing=%u kbps",
                   bitrate / 1000, next / 1000,
                   pacing_bitrate_for_encoder(next) / 1000);
    }
  } else if (ceiling > current) {
    my_zlog_debug("libdatachannel: REMB=%u kbps, ceiling %u kbps (ramping up)",
                  bitrate / 1000, ceiling / 1000);
  }
}

static void maybe_ramp_bitrate(RtcClient *client, int track, gint64 now_us)
{
  if (!client || track < 0)
    return;

  g_mutex_lock(&client->lock);
  if (now_us - client->last_bitrate_ramp_us <
      (gint64)WEBRTCPUSH_BITRATE_RAMP_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 sender_thread_main(gpointer data)
{
  RtcClient *client = data;
  const gint64 frame_us = 1000000 / WEBRTCPUSH_H264_FPS;
  const guint32 timestamp_step = 90000 / WEBRTCPUSH_H264_FPS;
  gint64 next_frame_us = 0;

  for (;;) {
    gboolean stopping;
    gboolean open;
    gboolean force_idr;
    int track;
    guint32 timestamp;
    const guint8 *frame;
    size_t frame_size;
    gboolean is_idr;
    guint64 source_pts = G_MAXUINT64;
    gint64 now;

    g_mutex_lock(&client->lock);
    stopping = client->stopping;
    open = client->track_open;
    track = client->track;
    force_idr = client->need_idr;
    timestamp = client->rtp_timestamp;
    g_mutex_unlock(&client->lock);

    if (stopping)
      break;
    if (!open || track < 0) {
      next_frame_us = 0;
      g_usleep(10000);
      continue;
    }

    now = g_get_monotonic_time();
    maybe_ramp_bitrate(client, track, now);
    if (next_frame_us == 0 || now - next_frame_us > frame_us * 4)
      next_frame_us = now;
    if (now < next_frame_us) {
      g_usleep((gulong)(next_frame_us - now));
      continue;
    }

    {
      gint pull_timeout = force_idr ? 4000 : 120;

      if (!mpp_h264_source_pull(client->h264_source, force_idr, &frame, &frame_size,
                                &is_idr, &source_pts, pull_timeout)) {
        my_zlog_warn("libdatachannel: live H264 frame pull timeout");
        next_frame_us = now + frame_us / 2;
        continue;
      }
    }

    g_mutex_lock(&client->lock);
    if (client->need_idr && !is_idr) {
      g_mutex_unlock(&client->lock);
      my_zlog_warn("libdatachannel: discard non-IDR while waiting for resync");
      next_frame_us = g_get_monotonic_time();
      continue;
    }
    if (source_pts != G_MAXUINT64) {
      if (!client->source_pts_valid || source_pts < client->last_source_pts) {
        client->source_pts_base = source_pts;
        client->last_source_pts = source_pts;
        client->rtp_timestamp_base = client->rtp_timestamp;
        client->source_pts_valid = TRUE;
      }
      timestamp = client->rtp_timestamp_base +
                  pts_delta_to_rtp(source_pts - client->source_pts_base);
      client->last_source_pts = source_pts;
    } 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, (int)frame_size) < 0) {
      g_mutex_unlock(&client->send_lock);
      my_zlog_warn("libdatachannel: rtcSendMessage failed");
      g_usleep(20000);
      continue;
    }
    g_mutex_unlock(&client->send_lock);

    g_mutex_lock(&client->lock);
    client->rtp_timestamp = timestamp + timestamp_step;
    if (is_idr)
      client->need_idr = FALSE;
    g_mutex_unlock(&client->lock);
    next_frame_us += frame_us;
  }
  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 gboolean parse_video_offer(const char *sdp, OfferVideo *video)
{
  gchar **lines;
  gboolean in_video = FALSE;
  gint mline = -1;
  gint i;

  memset(video, 0, sizeof(*video));
  video->payload_type = -1;
  video->mline_index = -1;
  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 ");
      if (in_video)
        video->mline_index = mline;
      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)
      continue;
    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;
    }
  }
  g_strfreev(lines);
  if (!video->mid && video->mline_index >= 0)
    video->mid = g_strdup_printf("%d", video->mline_index);
  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;

  g_mutex_lock(&client->lock);
  pc = client->pc;
  track = client->track;
  client->pc = -1;
  client->track = -1;
  client->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);

  g_mutex_lock(&client->send_lock);
  if (track >= 0) {
    rtcClose(track);
    rtcDeleteTrack(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";
  track_init.profile = "profile-level-id=42e01f;packetization-mode=1;level-asymmetry-allowed=1";
  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-video";
  packetizer.payloadType = (guint8)video->payload_type;
  packetizer.clockRate = 90000;
  packetizer.sequenceNumber = (guint16)g_random_int();
  packetizer.timestamp = g_random_int();
  packetizer.maxFragmentSize = 1150;
  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 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_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);

  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_free(client);
    return -1;
  }

  app->rtc_client = client;
  client->sender_thread = g_thread_new("rtc-h264-send", sender_thread_main, client);
  mpp_h264_source_set_bitrate(client->h264_source, client->target_bitrate);
  my_zlog_info("libdatachannel: MPP live source started device=%s mpp=%d fps=%d",
               video_device,
               mpp_h264_source_is_mpp_encoder(client->h264_source) ? 1 : 0,
               WEBRTCPUSH_H264_FPS);
  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_mutex_unlock(&client->lock);
  if (client->sender_thread)
    g_thread_join(client->sender_thread);

  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);
  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;
  rtcConfiguration config;
  int pc;
  int track = -1;
  gboolean duplicate;

  if (!client || !sdp)
    return -1;
  if (!parse_video_offer(sdp, &video)) {
    my_zlog_error("libdatachannel: phone offer has no usable H264 video");
    offer_video_clear(&video);
    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);
    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;
  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);
    my_zlog_error("libdatachannel: failed to create H264 track/handlers");
    return -1;
  }

  g_mutex_lock(&client->lock);
  client->pc = pc;
  client->track = 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_bitrate_ramp_us = g_get_monotonic_time();
  client->need_idr = TRUE;
  client->source_pts_valid = FALSE;
  g_mutex_unlock(&client->lock);

  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);
    destroy_peer(client);
    return -1;
  }
  flush_remote_ice(client);
  offer_video_clear(&video);
  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;
}
