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

#include <gst/gst.h>
#include <gst/app/gstappsrc.h>

/*
 * 手机→设备方向的音频播放管道（按键模式）：
 *   appsrc(opus payload) → opusdec → audioconvert → audioresample → volume → alsasink
 *
 * 手机端是"按住说话"模式（最长 15s），不是持续推流，因此：
 * - on_audio_message 收到包后入队，立即返回（不阻塞 libdatachannel 线程）
 * - 独立线程消费队列，维护"按键会话"
 * - 会话开始（首包到达）：pipeline 切到 PLAYING
 * - 会话结束（500ms 无包 或 15s 超时）：flush appsrc + pipeline 切到 READY
 *   切到 READY 释放 ALSA 设备，避免 alsasink 持续占用/空转
 */

#define AUDIO_SINK_SESSION_TIMEOUT_MS  500   /* 500ms 无包认为按键结束 */
#define AUDIO_SINK_SESSION_MAX_MS      15000 /* 单次按键最长 15s */

typedef struct {
  uint8_t *data;
  size_t size;
} SinkPacket;

struct AudioSink {
  GstElement *pipeline;
  GstElement *appsrc;
  GstElement *vol;

  GAsyncQueue *queue;          /* 待处理 Opus 包队列 */
  GMutex lock;                 /* 保护 pipeline 状态切换 */
  GThread *thread;             /* 消费线程 */
  gboolean quit;               /* 退出标志 */

  gint64 session_start_us;     /* 当前按键会话开始时间（0=无会话） */
  gint64 last_packet_us;       /* 最后一个包到达时间 */
  gboolean pipeline_playing;   /* pipeline 当前是否 PLAYING */
};

static void flush_queue(AudioSink *src)
{
  SinkPacket *pkt;
  while ((pkt = g_async_queue_try_pop(src->queue)) != NULL) {
    g_free(pkt->data);
    g_free(pkt);
  }
}

static void set_pipeline_state_locked(AudioSink *src, GstState state)
{
  if (!src->pipeline)
    return;
  /* 切到 READY 时 flush appsrc，避免旧数据残留导致下次会话首帧异常 */
  if (state == GST_STATE_READY && src->appsrc) {
    gst_element_send_event(src->pipeline,
        gst_event_new_flush_start());
    gst_element_send_event(src->pipeline,
        gst_event_new_flush_stop(FALSE));
  }
  gst_element_set_state(src->pipeline, state);
  src->pipeline_playing = (state == GST_STATE_PLAYING);
}

static gpointer audio_sink_thread(gpointer data)
{
  AudioSink *src = data;
  gint64 now;
  GstBuffer *buf;

  while (1) {
    /* 等待包，超时 100ms 用于检查会话超时 */
    SinkPacket *pkt = g_async_queue_timeout_pop(src->queue, 100 * 1000);
    now = g_get_monotonic_time();

    g_mutex_lock(&src->lock);
    if (src->quit) {
      g_mutex_unlock(&src->lock);
      if (pkt) { g_free(pkt->data); g_free(pkt); }
      break;
    }

    /* 会话超时检查 */
    if (src->session_start_us > 0) {
      gint64 idle_ms = (now - src->last_packet_us) / 1000;
      gint64 sess_ms = (now - src->session_start_us) / 1000;
      if (idle_ms >= AUDIO_SINK_SESSION_TIMEOUT_MS ||
          sess_ms >= AUDIO_SINK_SESSION_MAX_MS) {
        if (src->pipeline_playing) {
          set_pipeline_state_locked(src, GST_STATE_READY);
          my_zlog_info("audio_sink: session ended (idle=%lldms sess=%lldms)",
                       (long long)idle_ms, (long long)sess_ms);
        }
        src->session_start_us = 0;
        flush_queue(src);
        /* 丢弃超时后到达的旧包 */
        if (pkt) {
          g_free(pkt->data);
          g_free(pkt);
          pkt = NULL;
        }
        g_mutex_unlock(&src->lock);
        continue;
      }
    }

    if (pkt) {
      /* 新会话开始 */
      if (src->session_start_us == 0) {
        src->session_start_us = now;
        my_zlog_info("audio_sink: session start");
        if (!src->pipeline_playing)
          set_pipeline_state_locked(src, GST_STATE_PLAYING);
      }
      src->last_packet_us = now;

      /* push 到 appsrc（pipeline PLAYING 状态） */
      if (src->appsrc && src->pipeline_playing) {
        buf = gst_buffer_new_wrapped(g_memdup2(pkt->data, pkt->size), pkt->size);
        GST_BUFFER_DTS(buf) = GST_CLOCK_TIME_NONE;
        GST_BUFFER_PTS(buf) = GST_CLOCK_TIME_NONE;
        if (gst_app_src_push_buffer(GST_APP_SRC(src->appsrc), buf) != GST_FLOW_OK) {
          my_zlog_warn("audio_sink: push_buffer failed");
        }
      }
      g_free(pkt->data);
      g_free(pkt);
    }
    g_mutex_unlock(&src->lock);
  }

  return NULL;
}

AudioSink *audio_sink_start(const char *alsa_device, char **error_message)
{
  AudioSink *src;
  GstElement *pipe, *asrc, *dec, *conv, *resample, *vol, *sink;
  GstCaps *caps;
  GstStateChangeReturn ret;

  if (!alsa_device || !alsa_device[0]) {
    if (error_message)
      *error_message = g_strdup("no ALSA device");
    return NULL;
  }

  src = g_new0(AudioSink, 1);

  pipe = gst_pipeline_new("audio-sink-pipe");
  asrc = gst_element_factory_make("appsrc", "asrc");
  dec = gst_element_factory_make("opusdec", "dec");
  conv = gst_element_factory_make("audioconvert", "conv");
  resample = gst_element_factory_make("audioresample", "resample");
  vol = gst_element_factory_make("volume", "vol");
  sink = gst_element_factory_make("alsasink", "sink");

  if (!pipe || !asrc || !dec || !conv || !resample || !vol || !sink) {
    if (error_message)
      *error_message = g_strdup("failed to create audio sink GStreamer elements");
    if (pipe)
      gst_object_unref(pipe);
    g_free(src);
    return NULL;
  }

  caps = gst_caps_new_empty_simple("audio/x-opus");
  g_object_set(asrc,
      "caps", caps,
      "format", GST_FORMAT_BYTES,
      "is-live", TRUE,
      "emit-signals", FALSE,
      "min-latency", (gint64)0,
      "max-bytes", (guint64)(1 * 1024 * 1024),
      NULL);
  gst_caps_unref(caps);

  g_object_set(sink,
      "device", alsa_device,
      "buffer-time", (gint64)20000,
      "latency-time", (gint64)10000,
      "sync", FALSE,
      NULL);

  g_object_set(vol, "volume", 0.5, NULL);

  gst_bin_add_many(GST_BIN(pipe), asrc, dec, conv, resample, vol, sink, NULL);
  if (!gst_element_link_many(asrc, dec, conv, resample, vol, sink, NULL)) {
    if (error_message)
      *error_message = g_strdup("failed to link audio sink chain");
    gst_object_unref(pipe);
    g_free(src);
    return NULL;
  }

  src->pipeline = pipe;
  src->appsrc = asrc;
  src->vol = vol;
  src->queue = g_async_queue_new();
  g_mutex_init(&src->lock);
  src->session_start_us = 0;
  src->pipeline_playing = FALSE;

  /* 初始状态 READY（不占 ALSA 设备，等首包到来再 PLAYING） */
  ret = gst_element_set_state(pipe, GST_STATE_READY);
  if (ret == GST_STATE_CHANGE_FAILURE) {
    if (error_message)
      *error_message = g_strdup("audio sink pipeline failed to reach READY");
    gst_element_set_state(pipe, GST_STATE_NULL);
    gst_object_unref(pipe);
    g_async_queue_unref(src->queue);
    g_mutex_clear(&src->lock);
    g_free(src);
    return NULL;
  }

  src->thread = g_thread_new("audio-sink", audio_sink_thread, src);
  if (!src->thread) {
    if (error_message)
      *error_message = g_strdup("failed to create audio sink thread");
    audio_sink_stop(src);
    return NULL;
  }

  my_zlog_info("audio_sink: started device=%s opus=%uch %uHz (push-to-talk)",
      alsa_device, WEBRTCPUSH_OPUS_CHANNELS, WEBRTCPUSH_OPUS_CLOCKRATE);
  return src;
}

void audio_sink_stop(AudioSink *src)
{
  if (!src)
    return;
  if (src->thread) {
    g_mutex_lock(&src->lock);
    src->quit = TRUE;
    g_mutex_unlock(&src->lock);
    g_thread_join(src->thread);
    src->thread = NULL;
  }
  if (src->pipeline) {
    gst_element_set_state(src->pipeline, GST_STATE_NULL);
    gst_object_unref(src->pipeline);
  }
  if (src->queue) {
    flush_queue(src);
    g_async_queue_unref(src->queue);
  }
  g_mutex_clear(&src->lock);
  g_free(src);
}

gboolean audio_sink_push_opus(AudioSink *src, const uint8_t *data, size_t size)
{
  SinkPacket *pkt;
  if (!src || !src->queue || !data || size == 0)
    return FALSE;
  /* 入队，由消费线程处理（不阻塞 libdatachannel 回调线程） */
  pkt = g_new0(SinkPacket, 1);
  pkt->data = (uint8_t *)g_memdup2(data, size);
  pkt->size = size;
  g_async_queue_push(src->queue, pkt);
  return TRUE;
}

void audio_sink_set_volume(AudioSink *src, double volume)
{
  if (!src || !src->vol)
    return;
  if (volume < 0.0)
    volume = 0.0;
  if (volume > 1.0)
    volume = 1.0;
  g_object_set(src->vol, "volume", volume, NULL);
}
