use pts in gst buffer time.

server/pkg/gst/gst.c

@@ -95,10 +95,31 @@ static GstFlowReturn gstreamer_send_new_sample_handler(GstElement *object, gpoin
     buffer = gst_sample_get_buffer(sample);
     if (buffer) {
       gst_buffer_extract_dup(buffer, 0, gst_buffer_get_size(buffer), &copy, &copy_size);
-      goHandlePipelineBuffer(ctx->pipelineId, copy, copy_size,
+      
-        GST_BUFFER_DURATION(buffer),
+      // Get the buffer PTS (presentation timestamp)
-        GST_BUFFER_FLAG_IS_SET(buffer, GST_BUFFER_FLAG_DELTA_UNIT)
+      GstClockTime pts = GST_BUFFER_PTS(buffer);
-      );
+      
+      // Convert PTS to wallclock time by adding base time and subtracting pipeline time
+      GstClock *clock = gst_pipeline_get_clock(GST_PIPELINE(ctx->pipeline));
+      if (clock) {
+        gst_object_ref(clock);
+        GstClockTime base_time = gst_element_get_base_time(ctx->pipeline);
+        GstClockTime wallclock_time = pts + base_time;
+        gst_object_unref(clock);
+        
+        goHandlePipelineBuffer(ctx->pipelineId, copy, copy_size,
+          wallclock_time,
+          GST_BUFFER_DURATION(buffer),
+          GST_BUFFER_FLAG_IS_SET(buffer, GST_BUFFER_FLAG_DELTA_UNIT)
+        );
+      } else {
+        // Fallback to just PTS if clock is not available
+        goHandlePipelineBuffer(ctx->pipelineId, copy, copy_size,
+          pts,
+          GST_BUFFER_DURATION(buffer),
+          GST_BUFFER_FLAG_IS_SET(buffer, GST_BUFFER_FLAG_DELTA_UNIT)
+        );
+      }
     }
     gst_sample_unref(sample);
   }

server/pkg/gst/gst.go

@@ -200,7 +200,7 @@ func CheckPlugins(plugins []string) error {
 }
 
 //export goHandlePipelineBuffer
-func goHandlePipelineBuffer(pipelineID C.int, buf C.gpointer, bufLen C.int, duration C.guint64, deltaUnit C.gboolean) {
+func goHandlePipelineBuffer(pipelineID C.int, buf C.gpointer, bufLen C.int, pts C.guint64, duration C.guint64, deltaUnit C.gboolean) {
 	defer C.g_free(buf)
 
 	pipelinesLock.Lock()
@@ -208,10 +208,14 @@ func goHandlePipelineBuffer(pipelineID C.int, buf C.gpointer, bufLen C.int, dura
 	pipelinesLock.Unlock()
 
 	if ok {
+		// Convert GStreamer PTS (in nanoseconds since epoch) to Go time.Time
+		// GStreamer timestamps are in nanoseconds
+		timestamp := time.Unix(0, int64(pts))
+
 		pipeline.sample <- types.Sample{
 			Data:      C.GoBytes(unsafe.Pointer(buf), bufLen),
 			Length:    int(bufLen),
-			Timestamp: time.Now(),
+			Timestamp: timestamp,
 			Duration:  time.Duration(duration),
 			DeltaUnit: deltaUnit == C.TRUE,
 		}

server/pkg/gst/gst.h

@@ -23,7 +23,7 @@ typedef struct GstPipelineCtx {
   GstElement *appsrc;
 } GstPipelineCtx;
 
-extern void goHandlePipelineBuffer(int pipelineId, void *buffer, int bufferLen, guint64 duration, gboolean deltaUnit);
+extern void goHandlePipelineBuffer(int pipelineId, void *buffer, int bufferLen, guint64 pts, guint64 duration, gboolean deltaUnit);
 extern void goPipelineLog(int pipelineId, char *level, char *msg);
 
 GstPipelineCtx *gstreamer_pipeline_create(char *pipelineStr, int pipelineId, GError **error);