package handlers import ( "context" "fmt" "io" "os" "path/filepath" "strconv" "strings" "sync" "time" "visiona-agent/server/internal/api/ws" "visiona-agent/server/internal/camera" "visiona-agent/server/internal/device" "visiona-agent/server/internal/driver" "visiona-agent/server/internal/inference" "github.com/gin-gonic/gin" ) type CameraHandler struct { cameraMgr *camera.Manager deviceMgr *device.Manager inferenceSvc *inference.Service wsHub *ws.Hub streamer *camera.MJPEGStreamer pipeline *camera.InferencePipeline activeSource camera.FrameSource sourceType camera.SourceType // Video seek state — preserved across seek operations videoPath string // original file path videoFPS float64 // target FPS videoInfo camera.VideoInfo // duration, total frames activeDeviceID string // device ID for current video session // pendingStartCancel 取消「等 WS join room 才開跑 pipeline」的 gated-start goroutine // (video-inference-stuck 修法 A2)。stopActivePipeline 會呼叫它,確保下一次上傳 / // 停止時,還沒開跑的舊 pipeline 不會事後才 Start()(避免 pipeline 洩漏)。 pendingStartCancel context.CancelFunc // startMu 保護「gated-start 的 check-then-act」與 handler 端 stop 對 pendingStartCancel / // pipeline 的併發存取(Reviewer Major-1)。 // // 為什麼需要:gated goroutine「檢查 startCtx.Err() → pipeline.Start()」這兩步跨 goroutine // 非原子;若 handler goroutine 恰在中間呼叫 stopActivePipeline() → cancel(),舊 gated // goroutine 仍可能 Start 一個已被換掉的舊 pipeline。單一 Run() goroutine 只保護 Hub 內部, // 管不到 handler 端這段。用這把鎖把「二次檢查 + Start」原子化、並讓 stop 端的 // cancel + 換 pipeline 也在鎖內,兩者互斥。 // // 範圍刻意只涵蓋 pendingStartCancel / pipeline 這組跨 gated-goroutine 與 handler 的共享狀態, // 不擴大到 videoPath / activeSource 等其他欄位(維持原有請求序列化假設,避免無關擴大)。 startMu sync.Mutex } // waitRoomJoinTimeout 是 A2 gated-start 等待「結果 WS join inference room」的上限。 // // 逾時仍會開跑 pipeline(degrade 成舊行為),確保就算 WS 因故一直沒連上, // 影片推論也不會永久卡住(後續有 B 的 replay 緩存兜底早期結果)。 // 15s 足夠涵蓋 tunnel WS 握手 + 雲端 forward 的正常延遲。 const waitRoomJoinTimeout = 15 * time.Second func NewCameraHandler( cameraMgr *camera.Manager, deviceMgr *device.Manager, inferenceSvc *inference.Service, wsHub *ws.Hub, ) *CameraHandler { streamer := camera.NewMJPEGStreamer() go streamer.Run() return &CameraHandler{ cameraMgr: cameraMgr, deviceMgr: deviceMgr, inferenceSvc: inferenceSvc, wsHub: wsHub, streamer: streamer, } } func (h *CameraHandler) ListCameras(c *gin.Context) { cameras := h.cameraMgr.ListCameras() c.JSON(200, gin.H{"success": true, "data": gin.H{"cameras": cameras}}) } func (h *CameraHandler) StartPipeline(c *gin.Context) { var req struct { CameraID string `json:"cameraId"` DeviceID string `json:"deviceId"` Width int `json:"width"` Height int `json:"height"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": err.Error()}}) return } if req.Width == 0 { req.Width = 640 } if req.Height == 0 { req.Height = 480 } // Clean up any existing pipeline h.stopActivePipeline() // Open camera if err := h.cameraMgr.Open(0, req.Width, req.Height); err != nil { c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "CAMERA_OPEN_FAILED", "message": err.Error()}}) return } // Get device driver session, err := h.deviceMgr.GetDevice(req.DeviceID) if err != nil { c.JSON(404, gin.H{"success": false, "error": gin.H{"code": "DEVICE_NOT_FOUND", "message": err.Error()}}) return } // 新 pipeline 前清掉此 room 的 replay 緩存,避免上一次 session 的殘留結果補送給 client。 h.wsHub.ClearRoomReplay("inference:" + req.DeviceID) // Create inference result channel resultCh := make(chan *driver.InferenceResult, 10) // Forward results to WebSocket, enriching with device ID go func() { room := "inference:" + req.DeviceID for result := range resultCh { result.DeviceID = req.DeviceID h.wsHub.BroadcastToRoom(room, result) } }() // Start pipeline with camera as source h.activeSource = h.cameraMgr h.sourceType = camera.SourceCamera // Major-1:h.pipeline 由 stop 端在 startMu 內存取,這裡設定 + Start 也在鎖內保持一致。 h.startMu.Lock() h.pipeline = camera.NewInferencePipeline( h.cameraMgr, camera.SourceCamera, session.Driver, h.streamer.FrameChannel(), resultCh, ) h.pipeline.Start() h.startMu.Unlock() streamURL := "/api/camera/stream" c.JSON(200, gin.H{ "success": true, "data": gin.H{ "streamUrl": streamURL, "sourceType": "camera", }, }) } func (h *CameraHandler) StopPipeline(c *gin.Context) { h.stopActivePipeline() c.JSON(200, gin.H{"success": true}) } func (h *CameraHandler) StreamMJPEG(c *gin.Context) { h.streamer.ServeHTTP(c.Writer, c.Request) } // UploadImage handles image file upload for single-shot inference. func (h *CameraHandler) UploadImage(c *gin.Context) { h.stopActivePipeline() deviceID := c.PostForm("deviceId") if deviceID == "" { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "deviceId is required"}}) return } file, header, err := c.Request.FormFile("file") if err != nil { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "file is required"}}) return } defer file.Close() ext := strings.ToLower(filepath.Ext(header.Filename)) if ext != ".jpg" && ext != ".jpeg" && ext != ".png" { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "only JPG/PNG files are supported"}}) return } // Save to temp file tmpFile, err := os.CreateTemp("", "edge-ai-image-*"+ext) if err != nil { c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "STORAGE_ERROR", "message": err.Error()}}) return } if _, err := io.Copy(tmpFile, file); err != nil { tmpFile.Close() os.Remove(tmpFile.Name()) c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "STORAGE_ERROR", "message": err.Error()}}) return } tmpFile.Close() // Create ImageSource imgSource, err := camera.NewImageSource(tmpFile.Name()) if err != nil { os.Remove(tmpFile.Name()) c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "IMAGE_DECODE_FAILED", "message": err.Error()}}) return } // Get device driver session, err := h.deviceMgr.GetDevice(deviceID) if err != nil { imgSource.Close() c.JSON(404, gin.H{"success": false, "error": gin.H{"code": "DEVICE_NOT_FOUND", "message": err.Error()}}) return } // 新 pipeline 前清掉此 room 的 replay 緩存。image 只推論一次,replay 讓晚連的 WS // client 仍能補到那唯一一筆結果(順帶修 image 路徑同類的早期丟棄)。 h.wsHub.ClearRoomReplay("inference:" + deviceID) resultCh := make(chan *driver.InferenceResult, 10) go func() { room := "inference:" + deviceID for result := range resultCh { result.DeviceID = deviceID h.wsHub.BroadcastToRoom(room, result) } }() h.activeSource = imgSource h.sourceType = camera.SourceImage imgPipeline := camera.NewInferencePipeline( imgSource, camera.SourceImage, session.Driver, h.streamer.FrameChannel(), resultCh, ) // Major-1:h.pipeline 由 stop 端在 startMu 內存取,這裡設定 + Start 也在鎖內保持一致。 h.startMu.Lock() h.pipeline = imgPipeline imgPipeline.Start() h.startMu.Unlock() // Clean up result channel after pipeline completes // 用 local imgPipeline(非 h.pipeline)避免 goroutine 讀共享欄位。 go func() { <-imgPipeline.Done() close(resultCh) }() w, ht := imgSource.Dimensions() streamURL := "/api/camera/stream" c.JSON(200, gin.H{ "success": true, "data": gin.H{ "streamUrl": streamURL, "sourceType": "image", "width": w, "height": ht, "filename": header.Filename, }, }) } // UploadVideo handles video file upload for frame-by-frame inference. func (h *CameraHandler) UploadVideo(c *gin.Context) { h.stopActivePipeline() deviceID := c.PostForm("deviceId") if deviceID == "" { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "deviceId is required"}}) return } file, header, err := c.Request.FormFile("file") if err != nil { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "file is required"}}) return } defer file.Close() ext := strings.ToLower(filepath.Ext(header.Filename)) if ext != ".mp4" && ext != ".avi" && ext != ".mov" && ext != ".mpeg" && ext != ".mpg" { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "only MP4/AVI/MOV/MPEG/MPG files are supported"}}) return } // Save to temp file tmpFile, err := os.CreateTemp("", "edge-ai-video-*"+ext) if err != nil { c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "STORAGE_ERROR", "message": err.Error()}}) return } if _, err := io.Copy(tmpFile, file); err != nil { tmpFile.Close() os.Remove(tmpFile.Name()) c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "STORAGE_ERROR", "message": err.Error()}}) return } tmpFile.Close() // Probe video info (duration, frame count) before starting pipeline videoInfo := camera.ProbeVideoInfo(tmpFile.Name(), 15) // Create VideoSource videoSource, err := camera.NewVideoSource(tmpFile.Name(), 15) if err != nil { os.Remove(tmpFile.Name()) c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "VIDEO_DECODE_FAILED", "message": err.Error()}}) return } if videoInfo.TotalFrames > 0 { videoSource.SetTotalFrames(videoInfo.TotalFrames) } // Get device driver session, err := h.deviceMgr.GetDevice(deviceID) if err != nil { videoSource.Close() c.JSON(404, gin.H{"success": false, "error": gin.H{"code": "DEVICE_NOT_FOUND", "message": err.Error()}}) return } resultCh := make(chan *driver.InferenceResult, 10) go func() { room := "inference:" + deviceID for result := range resultCh { result.DeviceID = deviceID h.wsHub.BroadcastToRoom(room, result) } }() room := "inference:" + deviceID // 新一輪上傳:清掉舊的 replay 緩存,避免上一支影片的早期結果殘留補送給這次的 client。 h.wsHub.ClearRoomReplay(room) pipeline := camera.NewInferencePipeline( videoSource, camera.SourceVideo, session.Driver, h.streamer.FrameChannel(), resultCh, ) h.activeSource = videoSource h.sourceType = camera.SourceVideo h.videoPath = tmpFile.Name() h.videoFPS = 15 h.videoInfo = videoInfo h.activeDeviceID = deviceID // A2(主修):解耦「回 200」與「pipeline 開跑」。 // // 存檔完成即可回 200,但不立刻廣播推論結果——先在背景等結果 WS join // inference room,join 後(或逾時 degrade)才 pipeline.Start()。這樣影片 // 上傳走 localhost(極快)與結果訂閱走 tunnel WS(較慢)時序解耦後, // 早期結果不會在 Hub 因 room 無 client 被靜默丟棄(root cause §2)。 // // gated-start goroutine 用 startCtx 控制:stopActivePipeline 會 cancel 它, // 確保下一次上傳 / 停止時,這個還沒開跑的 pipeline 不會事後才 Start()。 // // Major-1:在 startMu 鎖內原子設定 pipeline + pendingStartCancel,讓後續可能併發的 // stop 看到一致的一對(pipeline 與其 cancel),不會讀到半設定狀態。 startCtx, cancelStart := context.WithCancel(context.Background()) h.startMu.Lock() h.pipeline = pipeline h.pendingStartCancel = cancelStart h.startMu.Unlock() go func() { waitCtx, waitCancel := context.WithTimeout(startCtx, waitRoomJoinTimeout) defer waitCancel() // 等到 room 有 client(true)或逾時(false, degrade 開跑)。 // startCtx 被 cancel(stopActivePipeline)→ WaitForRoomClient 回 false 且 // startCtx.Err()!=nil,此時不可開跑(pipeline 已被換掉 / 停止)。 _ = h.wsHub.WaitForRoomClient(waitCtx, room) // Major-1:把「二次檢查 startCtx.Err() → Start()」原子化。 // 取 startMu 後再檢查一次:若 stop 端已在等待與此刻之間 cancel 並換掉 pipeline, // startCtx.Err()!=nil,放棄開跑;否則在鎖內 Start,並清掉 pendingStartCancel // (已成功開跑,之後的 stop 改由 pipeline.Stop() 負責,不再靠 cancel)。 h.startMu.Lock() if startCtx.Err() != nil { h.startMu.Unlock() // 已被 stopActivePipeline 取消,放棄開跑。 // 必須關 resultCh,否則上面的 forwarder goroutine(range resultCh)永久阻塞洩漏。 // pipeline 從未 Start(),不會有人寫 resultCh,close 安全。 close(resultCh) return } pipeline.Start() // 這個 gated goroutine 的任務已完成:清掉自己登記的 cancel。 // startCtx.Err()==nil 保證期間沒有 stop 介入過(stop 會 cancel),故 pendingStartCancel // 必仍是自己登記的 cancelStart,直接清成 nil——之後的 stop 改由 pipeline.Stop() 負責。 h.pendingStartCancel = nil h.startMu.Unlock() // pipeline 跑完 → 關 resultCh、通知前端。放在開跑之後才註冊, // 避免「還沒 Start 就等 Done()」永久阻塞(NewInferencePipeline 的 doneCh 尚未 close)。 go func() { <-pipeline.Done() close(resultCh) h.wsHub.BroadcastToRoom(room, map[string]interface{}{ "type": "pipeline_complete", "sourceType": "video", }) }() }() streamURL := "/api/camera/stream" c.JSON(200, gin.H{ "success": true, "data": gin.H{ "streamUrl": streamURL, "sourceType": "video", "filename": header.Filename, "totalFrames": videoInfo.TotalFrames, "durationSeconds": videoInfo.DurationSec, }, }) } // UploadBatchImages handles multiple image files for sequential batch inference. func (h *CameraHandler) UploadBatchImages(c *gin.Context) { h.stopActivePipeline() deviceID := c.PostForm("deviceId") if deviceID == "" { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "deviceId is required"}}) return } form, err := c.MultipartForm() if err != nil { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "multipart form required"}}) return } files := form.File["files"] if len(files) == 0 { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "at least one file is required"}}) return } if len(files) > 50 { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "maximum 50 images per batch"}}) return } // Save all files to temp filePaths := make([]string, 0, len(files)) filenames := make([]string, 0, len(files)) for _, fh := range files { ext := strings.ToLower(filepath.Ext(fh.Filename)) if ext != ".jpg" && ext != ".jpeg" && ext != ".png" { for _, fp := range filePaths { os.Remove(fp) } c.JSON(400, gin.H{"success": false, "error": gin.H{ "code": "BAD_REQUEST", "message": fmt.Sprintf("unsupported file: %s (only JPG/PNG)", fh.Filename), }}) return } f, openErr := fh.Open() if openErr != nil { for _, fp := range filePaths { os.Remove(fp) } c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "STORAGE_ERROR", "message": openErr.Error()}}) return } tmpFile, tmpErr := os.CreateTemp("", "edge-ai-batch-*"+ext) if tmpErr != nil { f.Close() for _, fp := range filePaths { os.Remove(fp) } c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "STORAGE_ERROR", "message": tmpErr.Error()}}) return } io.Copy(tmpFile, f) tmpFile.Close() f.Close() filePaths = append(filePaths, tmpFile.Name()) filenames = append(filenames, fh.Filename) } // Create MultiImageSource batchSource, err := camera.NewMultiImageSource(filePaths, filenames) if err != nil { for _, fp := range filePaths { os.Remove(fp) } c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "IMAGE_DECODE_FAILED", "message": err.Error()}}) return } // Get device driver session, err := h.deviceMgr.GetDevice(deviceID) if err != nil { batchSource.Close() c.JSON(404, gin.H{"success": false, "error": gin.H{"code": "DEVICE_NOT_FOUND", "message": err.Error()}}) return } // 新 pipeline 前清掉此 room 的 replay 緩存(避免上一批殘留補送給 client)。 h.wsHub.ClearRoomReplay("inference:" + deviceID) batchID := fmt.Sprintf("batch-%d", time.Now().UnixNano()) resultCh := make(chan *driver.InferenceResult, 10) go func() { room := "inference:" + deviceID for result := range resultCh { result.DeviceID = deviceID h.wsHub.BroadcastToRoom(room, result) } }() h.activeSource = batchSource h.sourceType = camera.SourceBatchImage batchPipeline := camera.NewInferencePipeline( batchSource, camera.SourceBatchImage, session.Driver, h.streamer.FrameChannel(), resultCh, ) // Major-1:h.pipeline 由 stop 端在 startMu 內存取,這裡設定 + Start 也在鎖內保持一致。 h.startMu.Lock() h.pipeline = batchPipeline batchPipeline.Start() h.startMu.Unlock() // Notify frontend when batch completes // 用 local batchPipeline(非 h.pipeline)避免 goroutine 讀共享欄位。 go func() { <-batchPipeline.Done() close(resultCh) h.wsHub.BroadcastToRoom("inference:"+deviceID, map[string]interface{}{ "type": "pipeline_complete", "sourceType": "batch_image", "batchId": batchID, }) }() // Build image list for response imageList := make([]gin.H, len(batchSource.Images())) for i, entry := range batchSource.Images() { imageList[i] = gin.H{ "index": i, "filename": entry.Filename, "width": entry.Width, "height": entry.Height, } } streamURL := "/api/camera/stream" c.JSON(200, gin.H{ "success": true, "data": gin.H{ "streamUrl": streamURL, "sourceType": "batch_image", "batchId": batchID, "totalImages": len(files), "images": imageList, }, }) } // GetBatchImageFrame serves a specific image from the active batch by index. func (h *CameraHandler) GetBatchImageFrame(c *gin.Context) { if h.sourceType != camera.SourceBatchImage || h.activeSource == nil { c.JSON(404, gin.H{"success": false, "error": gin.H{"code": "NO_BATCH", "message": "no batch image source active"}}) return } indexStr := c.Param("index") index, err := strconv.Atoi(indexStr) if err != nil || index < 0 { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": "invalid index"}}) return } mis, ok := h.activeSource.(*camera.MultiImageSource) if !ok { c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "INTERNAL_ERROR", "message": "source type mismatch"}}) return } jpegData, err := mis.GetImageByIndex(index) if err != nil { c.JSON(404, gin.H{"success": false, "error": gin.H{"code": "NOT_FOUND", "message": err.Error()}}) return } c.Data(200, "image/jpeg", jpegData) } // cancelPendingStartAndStopPipeline 在 startMu 鎖內原子地: // 1. 取消尚未開跑的 gated-start goroutine(pendingStartCancel) // 2. Stop 並清掉 h.pipeline // // 這把鎖與 gated goroutine 的「二次檢查 + Start」共用,兩者互斥(Reviewer Major-1): // - 若此函式先取鎖:cancel startCtx + 清 pipeline → gated goroutine 之後取鎖時 // startCtx.Err()!=nil,放棄開跑。 // - 若 gated goroutine 先取鎖:Start 已完成、pendingStartCancel 已清 nil → 此函式的 // pipeline.Stop() 負責停掉已開跑的 pipeline。 // // 兩種情況都不會發生「stop 後 gated goroutine 又 Start 舊 pipeline」的洩漏。 func (h *CameraHandler) cancelPendingStartAndStopPipeline() { h.startMu.Lock() defer h.startMu.Unlock() if h.pendingStartCancel != nil { h.pendingStartCancel() h.pendingStartCancel = nil } if h.pipeline != nil { h.pipeline.Stop() h.pipeline = nil } } // stopPipelineForSeek stops the pipeline and ffmpeg process but keeps the video file. func (h *CameraHandler) stopPipelineForSeek() { // A2:seek 前也要原子地取消尚未開跑的 gated-start goroutine + 停 pipeline // (極端情況:上傳後 WS 還沒 join 就 seek)。cancel 後該 goroutine 自行 close 原 resultCh。 h.cancelPendingStartAndStopPipeline() if h.activeSource != nil { if vs, ok := h.activeSource.(*camera.VideoSource); ok { vs.CloseWithoutRemove() } } h.activeSource = nil } // stopActivePipeline stops the current pipeline and cleans up resources. func (h *CameraHandler) stopActivePipeline() { // A2 + Major-1:先原子地取消「等 WS join 才開跑」的 gated-start goroutine + 停 pipeline, // 確保尚未開跑的舊 pipeline 不會在此之後才 Start()。cancel 後該 goroutine 會自行 // close resultCh,不需在此處理。 h.cancelPendingStartAndStopPipeline() // 清掉 inference room 的 replay 緩存(若有 active 影片 session)。 if h.activeDeviceID != "" { h.wsHub.ClearRoomReplay("inference:" + h.activeDeviceID) } // Only close non-camera sources (camera is managed by cameraMgr) if h.activeSource != nil && h.sourceType != camera.SourceCamera { h.activeSource.Close() } if h.sourceType == camera.SourceCamera { h.cameraMgr.Close() } // ADR-019 §4.3.1 M1:補刪前一支影片的 temp 檔,防磁碟 DoS。 // // 為什麼要在這裡補:VideoSource.Close() 雖已 os.Remove(filePath),但 seek 流程用 // CloseWithoutRemove() 保留檔案供重新 seek,之後 h.videoPath 仍指向 temp 檔而 // activeSource 可能是不同的(或 nil)VideoSource。此處對 h.videoPath 明確補一次 // os.Remove 作為 belt-and-suspenders——已被刪過時第二次 Remove 是無害 no-op。 if h.videoPath != "" { _ = os.Remove(h.videoPath) } h.activeSource = nil h.sourceType = "" h.videoPath = "" h.activeDeviceID = "" } // SeekVideo seeks to a specific position in the current video and restarts inference. func (h *CameraHandler) SeekVideo(c *gin.Context) { var req struct { TimeSeconds float64 `json:"timeSeconds"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "BAD_REQUEST", "message": err.Error()}}) return } if h.videoPath == "" || h.sourceType != camera.SourceVideo { c.JSON(400, gin.H{"success": false, "error": gin.H{"code": "NO_VIDEO", "message": "no video is currently playing"}}) return } // Clamp seek time if req.TimeSeconds < 0 { req.TimeSeconds = 0 } if h.videoInfo.DurationSec > 0 && req.TimeSeconds > h.videoInfo.DurationSec { req.TimeSeconds = h.videoInfo.DurationSec } // Stop current pipeline without deleting the video file h.stopPipelineForSeek() // 清掉 seek 前的 replay 緩存,避免舊位置的結果被補送給 seek 後才 late-join 的 client。 // seek 不需 gated-start:WS client 早已 join(能觸發 seek 代表已在收結果),直接開跑。 h.wsHub.ClearRoomReplay("inference:" + h.activeDeviceID) // Create new VideoSource with seek position videoSource, err := camera.NewVideoSourceWithSeek(h.videoPath, h.videoFPS, req.TimeSeconds) if err != nil { c.JSON(500, gin.H{"success": false, "error": gin.H{"code": "SEEK_FAILED", "message": err.Error()}}) return } if h.videoInfo.TotalFrames > 0 { videoSource.SetTotalFrames(h.videoInfo.TotalFrames) } // Get device driver session, err := h.deviceMgr.GetDevice(h.activeDeviceID) if err != nil { videoSource.Close() c.JSON(404, gin.H{"success": false, "error": gin.H{"code": "DEVICE_NOT_FOUND", "message": err.Error()}}) return } // Calculate frame offset from seek position frameOffset := int(req.TimeSeconds * h.videoFPS) resultCh := make(chan *driver.InferenceResult, 10) go func() { room := "inference:" + h.activeDeviceID for result := range resultCh { result.DeviceID = h.activeDeviceID h.wsHub.BroadcastToRoom(room, result) } }() h.activeSource = videoSource seekPipeline := camera.NewInferencePipelineWithOffset( videoSource, camera.SourceVideo, session.Driver, h.streamer.FrameChannel(), resultCh, frameOffset, ) // Major-1:h.pipeline 由 stop 端在 startMu 內存取,這裡設定 + Start 也在鎖內保持一致。 // seek 不走 gated-start(WS 早已 join),故不設 pendingStartCancel。 h.startMu.Lock() h.pipeline = seekPipeline seekPipeline.Start() h.startMu.Unlock() // 用 local seekPipeline(非 h.pipeline)避免 goroutine 讀共享欄位。 go func() { <-seekPipeline.Done() close(resultCh) h.wsHub.BroadcastToRoom("inference:"+h.activeDeviceID, map[string]interface{}{ "type": "pipeline_complete", "sourceType": "video", }) }() c.JSON(200, gin.H{ "success": true, "data": gin.H{ "seekTo": req.TimeSeconds, "frameOffset": frameOffset, }, }) }