#include "ws_signaling.h"
#include "webrtcpush_run.h"
#include "webrtcpush_config.h"
#include "webrtcpush_log.h"
#include "rtc_client.h"

#include <json-glib/json-glib.h>
#include <libsoup/soup.h>
#include <string.h>

#include "gst_webrtc_pipeline.h"

static AppState *s_app;
static SoupSession *s_session;
static SoupWebsocketConnection *s_ws;
static GAsyncQueue *s_tx_queue;
static guint s_ping_id;
static guint s_reconnect_id;
static guint s_flush_id;
static gboolean s_ws_exiting;
static gchar *s_pending_offer_sdp;
static gchar *s_ws_url;
static GMainContext *s_main_ctx;
static gchar *s_debounced_offer_sdp = NULL;
static guint s_debounce_offer_id = 0;

static gboolean reconnect_cb(gpointer data);
static void ws_connect(void);
static void reset_signaling_session(AppState *app, gboolean teardown_pipeline);
static gboolean idle_pipeline_teardown(gpointer p);
static void schedule_tx_flush(void);
static void dispatch_json(const gchar *payload);

static void reset_signaling_session(AppState *app, gboolean teardown_pipeline) {
  if (!app)
    return;
  g_free(app->member_id);
  app->member_id = NULL;
  if (teardown_pipeline)
    app->call_enable = FALSE;
  g_free(s_pending_offer_sdp);
  s_pending_offer_sdp = NULL;
  if (teardown_pipeline)
    g_idle_add(idle_pipeline_teardown, app);
}

typedef struct {
  AppState *app;
  gchar *sdp;
} OfferIdle;

static gboolean tx_flush_tick(gpointer data) {
  (void)data;
  s_flush_id = 0;
  if (!s_ws || soup_websocket_connection_get_state(s_ws) != SOUP_WEBSOCKET_STATE_OPEN)
    return G_SOURCE_REMOVE;

  gchar *msg = g_async_queue_try_pop(s_tx_queue);
  if (!msg)
    return G_SOURCE_REMOVE;

  soup_websocket_connection_send_text(s_ws, msg);
  if (s_app && s_app->verbose)
    g_print("[verbose] WS tx %zu bytes\n", strlen(msg));
  g_free(msg);

  if (g_async_queue_length(s_tx_queue) > 0)
    s_flush_id = g_timeout_add(15, tx_flush_tick, NULL);
  return G_SOURCE_REMOVE;
}

static gboolean idle_schedule_tx_flush(gpointer data) {
  (void)data;
  if (!s_flush_id)
    s_flush_id = g_timeout_add(0, tx_flush_tick, NULL);
  return G_SOURCE_REMOVE;
}

/* GStreamer ICE/SDP 回调可能在 streaming 线程；g_timeout_add 必须在主线程 */
static void schedule_tx_flush(void) {
  if (!s_main_ctx) {
    idle_schedule_tx_flush(NULL);
    return;
  }
  g_main_context_invoke(s_main_ctx, idle_schedule_tx_flush, NULL);
}

static void queue_send(const gchar *text) {
  if (!text || !text[0])
    return;
  g_async_queue_push(s_tx_queue, g_strdup(text));
  schedule_tx_flush();
}

void ws_send_text(AppState *app, const char *text) {
  (void)app;
  queue_send(text);
}

gboolean ws_signaling_ready(void) {
  return !s_ws_exiting && s_ws &&
         soup_websocket_connection_get_state(s_ws) == SOUP_WEBSOCKET_STATE_OPEN;
}

void ws_signaling_service_burst(unsigned n_ticks) {
  (void)n_ticks;
  schedule_tx_flush();
}

static gboolean ping_tick(gpointer data) {
  (void)data;
  queue_send("{\"type\":\"ping\",\"message\":\"keeplive\"}");
  return G_SOURCE_CONTINUE;
}

static void schedule_offer(AppState *app, const gchar *sdp);

typedef struct {
  AppState *app;
  gchar *sdp;
} AnswerIdle;

static gboolean idle_answer(gpointer user_data) {
  AnswerIdle *a = user_data;
  gst_webrtc_handle_remote_answer(a->app, a->sdp);
  g_free(a->sdp);
  g_free(a);
  return G_SOURCE_REMOVE;
}

static void schedule_answer(AppState *app, const gchar *sdp) {
  if (!app || !sdp || !app->device_offerer)
    return;
  AnswerIdle *a = g_new(AnswerIdle, 1);
  a->app = app;
  a->sdp = g_strdup(sdp);
  g_idle_add(idle_answer, a);
}

static gboolean idle_apply_remote_offer(gpointer user_data) {
  OfferIdle *o = user_data;
  if (!o->app->pipeline || !o->app->webrtc) {
    g_printerr("Pipeline gone before set-remote\n");
    g_free(o->sdp);
    g_free(o);
    return G_SOURCE_REMOVE;
  }
  gst_webrtc_pipeline_apply_remote_offer(o->app, o->sdp);
  g_free(o->sdp);
  g_free(o);
  return G_SOURCE_REMOVE;
}

static gboolean idle_offer(gpointer user_data) {
  OfferIdle *o = user_data;
  AppState *app = o->app;

  if (rtc_client_handle_offer(app, o->sdp) != 0)
    my_zlog_error("libdatachannel: failed to apply phone offer");
  g_free(o->sdp);
  g_free(o);
  return G_SOURCE_REMOVE;

#if 0 /* Kept in gst_webrtc_pipeline.c; native branch now uses libdatachannel. */

  if (app->pipeline && gst_webrtc_pipeline_should_ignore_phone_offer(app, o->sdp)) {
    gst_webrtc_pipeline_debug_log_state(app, "ignore-offer");
    my_zlog_info("webrtcpush: ignore phone offer (duplicate ufrag / negotiating)");
    g_free(o->sdp);
    g_free(o);
    return G_SOURCE_REMOVE;
  }

  if (gst_webrtc_pipeline_can_renegotiate(app, o->sdp)) {
    gst_webrtc_pipeline_debug_log_state(app, "renegotiate");
    gst_webrtc_pipeline_renegotiate_remote_offer(app, o->sdp);
    g_free(o->sdp);
    g_free(o);
    return G_SOURCE_REMOVE;
  }

  /* 已 link 的会话禁止 teardown（手机重连只允许 renegotiate） */
  if (app->pipeline && app->payloader_linked) {
    my_zlog_warn("webrtcpush: linked session, skip offer (no rebuild)");
    g_free(o->sdp);
    g_free(o);
    return G_SOURCE_REMOVE;
  }

  /* 预热管道 / 首包 Offer：直接 set-remote，禁止 teardown+rebuild */
  if (app->pipeline && !app->remote_description_set) {
    my_zlog_info("webrtcpush: apply first offer on existing pipeline (no rebuild)");
    g_idle_add(idle_apply_remote_offer, o);
    return G_SOURCE_REMOVE;
  }

  if (app->pipeline)
    gst_webrtc_pipeline_teardown(app);

  if (!gst_webrtc_pipeline_build(app)) {
    g_printerr("Failed to build GStreamer pipeline\n");
    webrtcpush_notify_fatal_failure("GStreamer pipeline build failed");
    g_free(o->sdp);
    g_free(o);
    return G_SOURCE_REMOVE;
  }

  g_idle_add(idle_apply_remote_offer, o);
  return G_SOURCE_REMOVE;
#endif
}

static gboolean debounced_offer_fire(gpointer user_data) {
  AppState *app = user_data;
  OfferIdle *o;

  s_debounce_offer_id = 0;
  if (!s_debounced_offer_sdp || !app)
    return G_SOURCE_REMOVE;

  o = g_new(OfferIdle, 1);
  o->app = app;
  o->sdp = s_debounced_offer_sdp;
  s_debounced_offer_sdp = NULL;
  g_idle_add(idle_offer, o);
  return G_SOURCE_REMOVE;
}

static void schedule_offer(AppState *app, const gchar *sdp) {
  if (!app->member_id) {
    g_free(s_pending_offer_sdp);
    s_pending_offer_sdp = g_strdup(sdp);
    g_print("Offer received before id; holding SDP\n");
    return;
  }
  g_free(s_debounced_offer_sdp);
  s_debounced_offer_sdp = g_strdup(sdp);
  if (s_debounce_offer_id)
    g_source_remove(s_debounce_offer_id);
  s_debounce_offer_id = g_timeout_add(40, debounced_offer_fire, app);
  return;

#if 0 /* Old webrtcbin negotiation is intentionally retained but not selected. */
  /* client.html 每秒发 Offer；若本端已发出 Offer 则忽略，否则应答手机 Offer */
  if (app->device_offerer && gst_webrtc_pipeline_offer_in_flight(app)) {
    g_print("Offer-mode: ignore phone offer (local offer sent, waiting for answer)\n");
    return;
  }
  if (app->device_offerer) {
    g_print("Offer-mode: phone offer first, answering (client.html glare)\n");
    gst_webrtc_pipeline_stop_offer_retry();
  }
  gst_webrtc_pipeline_on_rx_offer_sdp(app, sdp);
  if (!app->member_id) {
    g_free(s_pending_offer_sdp);
    s_pending_offer_sdp = g_strdup(sdp);
    g_print("Offer received before id; holding SDP\n");
    return;
  }
  g_free(s_debounced_offer_sdp);
  s_debounced_offer_sdp = g_strdup(sdp);
  if (s_debounce_offer_id) {
    g_source_remove(s_debounce_offer_id);
    s_debounce_offer_id = 0;
  }
  /* 安卓进驾驶/换驾驶员：新 ice-ufrag，立即应答，不等 80ms 去抖 */
  {
    guint delay_ms =
        gst_webrtc_pipeline_is_phone_reconnect_offer(app, sdp) ? 0 : 80;
    s_debounce_offer_id = g_timeout_add(delay_ms, debounced_offer_fire, app);
  }
#endif
}

static void flush_pending_offer(void) {
  if (!s_pending_offer_sdp || !s_app->member_id)
    return;
  gchar *sdp = s_pending_offer_sdp;
  s_pending_offer_sdp = NULL;
  schedule_offer(s_app, sdp);
  g_free(sdp);
}

static gboolean idle_ice(gpointer user_data) {
  JsonObject *obj = user_data;
  rtc_client_handle_remote_ice(s_app, obj);
  json_object_unref(obj);
  return G_SOURCE_REMOVE;
}

static void dispatch_json(const gchar *payload) {
  if (s_app && s_app->verbose)
    g_print("[verbose] rx: %s\n", payload);

  if (!payload || !payload[0])
    return;
  if (!strcmp(payload, "null"))
    return;

  JsonParser *parser = json_parser_new();
  GError *err = NULL;
  if (!json_parser_load_from_data(parser, payload, -1, &err)) {
    g_printerr("JSON parse error: %s\n", err ? err->message : "?");
    if (err)
      g_error_free(err);
    g_object_unref(parser);
    return;
  }

  JsonNode *root_node = json_parser_get_root(parser);
  if (!root_node || JSON_NODE_HOLDS_NULL(root_node)) {
    g_object_unref(parser);
    return;
  }
  if (!JSON_NODE_HOLDS_OBJECT(root_node)) {
    g_object_unref(parser);
    return;
  }

  JsonObject *root = json_node_get_object(root_node);

  if (json_object_has_member(root, "type")) {
    const gchar *t = json_object_get_string_member(root, "type");

    if (!g_strcmp0(t, "id")) {
      if (json_object_has_member(root, "message")) {
        JsonNode *mn = json_object_get_member(root, "message");
        if (JSON_NODE_HOLDS_VALUE(mn) && json_node_get_value_type(mn) == G_TYPE_INT64) {
          gint64 v = json_node_get_int(mn);
          g_free(s_app->member_id);
          s_app->member_id = g_strdup_printf("%" G_GINT64_FORMAT, v);
        } else if (json_node_get_value_type(mn) == G_TYPE_STRING) {
          g_free(s_app->member_id);
          s_app->member_id = g_strdup(json_node_get_string(mn));
        }
        g_print("memberId=%s\n", s_app->member_id ? s_app->member_id : "?");
        if (s_app->device_offerer) {
          if (s_app->call_enable) {
            gst_webrtc_pipeline_start_offer_retry(s_app);
            gst_webrtc_pipeline_try_start_offer(s_app);
          }
        } else {
          flush_pending_offer();
        }
      }
      g_object_unref(parser);
      return;
    }

    if (!g_strcmp0(t, "num")) {
      if (json_object_has_member(root, "message")) {
        gint64 n = json_object_get_int_member(root, "message");
        g_print("num peers: %" G_GINT64_FORMAT "\n", n);
        if (n < 2)
          g_print("  (waiting for phone: https://jywy.yd-ss.com/client.html?dev=%s)\n",
                  s_app->dev_room ? s_app->dev_room : "?");
      }
      g_object_unref(parser);
      return;
    }

    if (!g_strcmp0(t, "answer")) {
      if (s_app->device_offerer) {
        const gchar *sdp = json_object_get_string_member(root, "sdp");
        if (sdp)
          schedule_answer(s_app, sdp);
      }
      g_object_unref(parser);
      return;
    }

    if (!g_strcmp0(t, "offer")) {
      const gchar *sdp = json_object_get_string_member(root, "sdp");
      if (sdp)
        schedule_offer(s_app, sdp);
      g_object_unref(parser);
      return;
    }

    if (!g_strcmp0(t, "pong")) {
      g_object_unref(parser);
      return;
    }

    if (!g_strcmp0(t, "leave")) {
      if (rtc_client_is_active(s_app)) {
        g_print("peer left: reset libdatachannel peer and wait for a new offer\n");
        g_idle_add(idle_pipeline_teardown, s_app);
      }
      g_object_unref(parser);
      return;
    }
  }

  if (json_object_has_member(root, "description")) {
    JsonNode *dn = json_object_get_member(root, "description");
    if (JSON_NODE_HOLDS_OBJECT(dn)) {
      JsonObject *desc = json_node_get_object(dn);
      if (json_object_has_member(desc, "type")) {
        const gchar *dtype = json_object_get_string_member(desc, "type");
        if (!g_strcmp0(dtype, "offer")) {
          const gchar *sdp = json_object_get_string_member(desc, "sdp");
          if (sdp)
            schedule_offer(s_app, sdp);
          g_object_unref(parser);
          return;
        }
        if (s_app->device_offerer && !g_strcmp0(dtype, "answer")) {
          const gchar *sdp = json_object_get_string_member(desc, "sdp");
          if (sdp)
            schedule_answer(s_app, sdp);
          g_object_unref(parser);
          return;
        }
      }
    }
  }

  if (json_object_has_member(root, "candidate")) {
    JsonNode *cn = json_object_get_member(root, "candidate");
    if (JSON_NODE_HOLDS_OBJECT(cn)) {
      JsonObject *co = json_node_get_object(cn);
      json_object_ref(co);
      g_idle_add(idle_ice, co);
    } else if (JSON_NODE_HOLDS_VALUE(cn) && json_node_get_value_type(cn) == G_TYPE_STRING) {
      JsonObject *co = json_object_new();
      json_object_set_string_member(co, "candidate", json_node_get_string(cn));
      if (json_object_has_member(root, "sdpMLineIndex"))
        json_object_set_int_member(co, "sdpMLineIndex", json_object_get_int_member(root, "sdpMLineIndex"));
      if (json_object_has_member(root, "sdpMid")) {
        JsonNode *midn = json_object_get_member(root, "sdpMid");
        if (json_node_get_value_type(midn) == G_TYPE_STRING)
          json_object_set_string_member(co, "sdpMid", json_object_get_string_member(root, "sdpMid"));
      }
      g_idle_add(idle_ice, co);
    }
  }

  g_object_unref(parser);
}

static void on_ws_message(SoupWebsocketConnection *conn, SoupWebsocketDataType type,
                          GBytes *message, gpointer user_data) {
  (void)conn;
  (void)user_data;
  if (type != SOUP_WEBSOCKET_DATA_TEXT)
    return;
  gsize len = 0;
  const gchar *data = g_bytes_get_data(message, &len);
  if (!data || len == 0)
    return;
  gchar *payload = g_strndup(data, len);
  g_strstrip(payload);
  if (payload[0])
    dispatch_json(payload);
  g_free(payload);
}

static void on_ws_closed(SoupWebsocketConnection *conn, gpointer user_data) {
  (void)user_data;
  guint code = conn ? soup_websocket_connection_get_close_code(conn) : 0;
  gboolean keep_pipeline = s_app && rtc_client_media_open(s_app);

  g_print("WebSocket closed code=%u; reconnecting signaling", code);
  if (s_ws) {
    g_object_unref(s_ws);
    s_ws = NULL;
  }
  if (!s_ws_exiting && s_app) {
    if (keep_pipeline)
      g_print(" (keep WebRTC pipeline)\n");
    else if (rtc_client_is_active(s_app))
      g_print(" (reset libdatachannel peer)\n");
    else
      g_print("\n");
    reset_signaling_session(s_app, !keep_pipeline);
    if (s_reconnect_id)
      g_source_remove(s_reconnect_id);
    s_reconnect_id = g_timeout_add_seconds(2, reconnect_cb, NULL);
  }
}

static void on_ws_connected(GObject *source_object, GAsyncResult *res, gpointer user_data) {
  (void)user_data;
  GError *err = NULL;
  s_ws = soup_session_websocket_connect_finish(SOUP_SESSION(source_object), res, &err);
  if (err) {
    g_printerr("WebSocket connect failed: %s\n", err->message);
    g_error_free(err);
    if (!s_ws_exiting && !s_reconnect_id)
      s_reconnect_id = g_timeout_add_seconds(2, reconnect_cb, NULL);
    return;
  }
  g_signal_connect(s_ws, "message", G_CALLBACK(on_ws_message), NULL);
  g_signal_connect(s_ws, "closed", G_CALLBACK(on_ws_closed), NULL);
  g_print("WebSocket connected (libsoup)\n");
  schedule_tx_flush();
}

static gboolean idle_pipeline_teardown(gpointer p) {
  rtc_client_reset((AppState *)p);
  return G_SOURCE_REMOVE;
}

static void ws_connect(void) {
  if (s_ws_exiting || !s_app || !s_ws_url)
    return;
  if (!s_session)
    s_session = soup_session_new();

  SoupMessage *msg = soup_message_new("GET", s_ws_url);
  if (!msg) {
    g_printerr("soup_message_new failed\n");
    return;
  }
  soup_message_headers_append(msg->request_headers, "Origin", "https://jywy.yd-ss.com");
  soup_session_websocket_connect_async(s_session, msg, NULL, NULL, NULL, on_ws_connected, NULL);
  g_object_unref(msg);
}

static gboolean reconnect_cb(gpointer data) {
  (void)data;
  s_reconnect_id = 0;
  if (s_ws_exiting || !s_app)
    return G_SOURCE_REMOVE;
  g_print("Reconnecting WebSocket...\n");
  ws_connect();
  return G_SOURCE_REMOVE;
}

int ws_signaling_start(AppState *app) {
  s_app = app;
  s_ws_exiting = FALSE;

  if (!s_main_ctx)
    s_main_ctx = g_main_context_ref(g_main_context_default());

  if (!s_tx_queue)
    s_tx_queue = g_async_queue_new();

  g_free(s_ws_url);
  s_ws_url = g_strdup_printf("wss://signal.yd-ss.com/websocket?dev=%s", app->dev_room);

  if (!s_ping_id)
    s_ping_id = g_timeout_add_seconds(3, ping_tick, NULL);

  ws_connect();
  return 0;
}

static gboolean idle_signaling_refresh(gpointer user_data) {
  AppState *app = user_data;
  if (!app)
    return G_SOURCE_REMOVE;

  rtc_client_reset(app);
  g_print("signaling refresh: libdatachannel peer reset (await phone offer)\n");

  if (s_ws && soup_websocket_connection_get_state(s_ws) == SOUP_WEBSOCKET_STATE_OPEN) {
    g_print("signaling refresh: reconnect WebSocket\n");
    soup_websocket_connection_close(s_ws, SOUP_WEBSOCKET_CLOSE_NORMAL, NULL);
  } else if (!s_ws_exiting) {
    reset_signaling_session(app, FALSE);
    ws_connect();
  }
  return G_SOURCE_REMOVE;
}

void ws_signaling_request_refresh(AppState *app) {
  if (!app || s_ws_exiting)
    return;
  if (s_main_ctx)
    g_main_context_invoke(s_main_ctx, idle_signaling_refresh, app);
  else
    g_idle_add(idle_signaling_refresh, app);
}

static gboolean idle_signaling_soft_refresh(gpointer user_data) {
  AppState *app = user_data;
  if (!app)
    return G_SOURCE_REMOVE;

  rtc_client_reset(app);
  my_zlog_info("webrtcpush soft refresh: libdatachannel peer reset, WS kept");
  return G_SOURCE_REMOVE;
}

void ws_signaling_request_soft_refresh(AppState *app) {
  if (!app || s_ws_exiting)
    return;
  if (s_main_ctx)
    g_main_context_invoke(s_main_ctx, idle_signaling_soft_refresh, app);
  else
    g_idle_add(idle_signaling_soft_refresh, app);
}

void ws_signaling_stop(AppState *app) {
  (void)app;
  s_ws_exiting = TRUE;

  if (s_flush_id) {
    g_source_remove(s_flush_id);
    s_flush_id = 0;
  }
  if (s_reconnect_id) {
    g_source_remove(s_reconnect_id);
    s_reconnect_id = 0;
  }
  if (s_ping_id) {
    g_source_remove(s_ping_id);
    s_ping_id = 0;
  }

  if (s_ws) {
    soup_websocket_connection_close(s_ws, SOUP_WEBSOCKET_CLOSE_NORMAL, NULL);
    g_object_unref(s_ws);
    s_ws = NULL;
  }
  if (s_session) {
    g_object_unref(s_session);
    s_session = NULL;
  }

  g_free(s_ws_url);
  s_ws_url = NULL;
  g_free(s_pending_offer_sdp);
  s_pending_offer_sdp = NULL;

  if (s_tx_queue) {
    gchar *m;
    while ((m = g_async_queue_try_pop(s_tx_queue)) != NULL)
      g_free(m);
    g_async_queue_unref(s_tx_queue);
    s_tx_queue = NULL;
  }

  if (s_main_ctx) {
    g_main_context_unref(s_main_ctx);
    s_main_ctx = NULL;
  }
}
