feat: implement Replay/Time-Travel REST endpoints
Add OpenAPI-style godoc comments and comprehensive table-driven tests for replay endpoints: - GET /api/replay/sessions - list recording sessions and replay store info - POST /api/replay/start - start replay at timestamp (speed 1/2/5) - POST /api/replay/stop - stop replay, return to live - POST /api/replay/seek - seek within session - POST /api/replay/tune - update pipeline parameters mid-replay Improvements: - Fix writeJSON calls to use proper 3-argument signature - Add detailed request/response type documentation - Add mockRecordingStore for isolated unit testing - Add 12 table-driven test cases covering all endpoints Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jedarden
parent
12efb9a097
commit
a873663cfc
2 changed files with 1267 additions and 39 deletions
|
|
@ -65,11 +65,71 @@ func (h *ReplayHandler) Close() error {
|
|||
|
||||
// RegisterRoutes registers replay endpoints.
|
||||
//
|
||||
// GET /api/replay/sessions — list available recording sessions
|
||||
// POST /api/replay/start — start replay at given timestamp
|
||||
// POST /api/replay/stop — stop replay, return to live
|
||||
// POST /api/replay/seek — seek to timestamp within session
|
||||
// POST /api/replay/tune — update pipeline parameters mid-replay
|
||||
// Replay/Time-Travel Endpoints:
|
||||
//
|
||||
// GET /api/replay/sessions — list recording sessions and replay store info
|
||||
//
|
||||
// @Summary List replay sessions
|
||||
// @Description Returns information about available recorded data and active replay sessions.
|
||||
// @Description Includes file size, timestamp range, and all active sessions.
|
||||
// @Tags replay
|
||||
// @Produce json
|
||||
// @Success 200 {object} replayInfo "Replay store info and active sessions"
|
||||
// @Router /api/replay/sessions [get]
|
||||
//
|
||||
// POST /api/replay/start — start replay at given timestamp
|
||||
//
|
||||
// @Summary Start replay session
|
||||
// @Description Creates a new replay session for the specified time range. The session
|
||||
// @Description starts in paused state. Use speed to control playback rate (1, 2, or 5).
|
||||
// @Tags replay
|
||||
// @Accept json
|
||||
// @Produce json
|
||||
// @Param request body startSessionRequest true "Replay start parameters"
|
||||
// @Success 200 {object} map[string]interface{} "Session created with ID and state"
|
||||
// @Failure 400 {object} map[string]string "Invalid request parameters"
|
||||
// @Router /api/replay/start [post]
|
||||
//
|
||||
// POST /api/replay/stop — stop replay, return to live
|
||||
//
|
||||
// @Summary Stop replay session
|
||||
// @Description Stops the specified replay session and returns to live mode.
|
||||
// @Tags replay
|
||||
// @Accept json
|
||||
// @Produce json
|
||||
// @Param request body stopSessionRequest true "Session to stop"
|
||||
// @Success 200 {object} map[string]string "Session stopped"
|
||||
// @Failure 404 {object} map[string]string "Session not found"
|
||||
// @Router /api/replay/stop [post]
|
||||
//
|
||||
// POST /api/replay/seek — seek to timestamp within session
|
||||
//
|
||||
// @Summary Seek within replay session
|
||||
// @Description Moves the replay cursor to the specified timestamp within the session range.
|
||||
// @Description Pauses playback and reads one frame at the target position.
|
||||
// @Tags replay
|
||||
// @Accept json
|
||||
// @Produce json
|
||||
// @Param request body seekRequest true "Seek parameters"
|
||||
// @Success 200 {object} map[string]interface{} "Seek complete with current position"
|
||||
// @Failure 400 {object} map[string]string "Invalid timestamp or out of range"
|
||||
// @Failure 404 {object} map[string]string "Session not found"
|
||||
// @Router /api/replay/seek [post]
|
||||
//
|
||||
// POST /api/replay/tune — update pipeline parameters mid-replay
|
||||
//
|
||||
// @Summary Tune replay pipeline parameters
|
||||
// @Description Updates detection pipeline parameters for the replay session without
|
||||
// @Description affecting live processing. Useful for exploring how parameter changes
|
||||
// @Description affect detection on historical data.
|
||||
// @Tags replay
|
||||
// @Accept json
|
||||
// @Produce json
|
||||
// @Param request body tuneRequest true "Parameter updates"
|
||||
// @Success 200 {object} map[string]interface{} "Parameters updated"
|
||||
// @Failure 400 {object} map[string]string "Invalid request"
|
||||
// @Failure 404 {object} map[string]string "Session not found"
|
||||
// @Router /api/replay/tune [post]
|
||||
func (h *ReplayHandler) RegisterRoutes(r chi.Router) {
|
||||
r.Get("/api/replay/sessions", h.listSessions)
|
||||
r.Post("/api/replay/start", h.startSession)
|
||||
|
|
@ -78,16 +138,19 @@ func (h *ReplayHandler) RegisterRoutes(r chi.Router) {
|
|||
r.Post("/api/replay/tune", h.tune)
|
||||
}
|
||||
|
||||
// replayInfo represents the response from GET /api/replay/sessions.
|
||||
type replayInfo struct {
|
||||
HasData bool `json:"has_data"`
|
||||
FileSize int64 `json:"file_size_mb"`
|
||||
WritePos int64 `json:"write_pos"`
|
||||
OldestPos int64 `json:"oldest_pos"`
|
||||
OldestTS int64 `json:"oldest_timestamp_ms"`
|
||||
NewestTS int64 `json:"newest_timestamp_ms"`
|
||||
HasData bool `json:"has_data"`
|
||||
FileSize int64 `json:"file_size_mb"`
|
||||
WritePos int64 `json:"write_pos"`
|
||||
OldestPos int64 `json:"oldest_pos"`
|
||||
OldestTS int64 `json:"oldest_timestamp_ms"`
|
||||
NewestTS int64 `json:"newest_timestamp_ms"`
|
||||
Sessions []*_replaySession `json:"sessions"`
|
||||
}
|
||||
|
||||
// listSessions handles GET /api/replay/sessions.
|
||||
// Returns replay store statistics and all active sessions.
|
||||
func (h *ReplayHandler) listSessions(w http.ResponseWriter, r *http.Request) {
|
||||
stats := h.store.Stats()
|
||||
|
||||
|
|
@ -115,25 +178,31 @@ func (h *ReplayHandler) listSessions(w http.ResponseWriter, r *http.Request) {
|
|||
Sessions: sessions,
|
||||
}
|
||||
|
||||
writeJSON(w, info)
|
||||
writeJSON(w, http.StatusOK, info)
|
||||
}
|
||||
|
||||
// startSessionRequest represents the request body for POST /api/replay/start.
|
||||
type startSessionRequest struct {
|
||||
// FromISO8601 is the start timestamp in ISO8601 format (e.g., "2024-03-15T14:30:00Z")
|
||||
FromISO8601 string `json:"from_iso8601"`
|
||||
ToISO8601 string `json:"to_iso8601"`
|
||||
Speed int `json:"speed,omitempty"` // 1, 2, 5
|
||||
// ToISO8601 is the end timestamp in ISO8601 format. If empty, defaults to now.
|
||||
ToISO8601 string `json:"to_iso8601"`
|
||||
// Speed is the playback speed multiplier: 1, 2, or 5. Defaults to 1.
|
||||
Speed int `json:"speed,omitempty"`
|
||||
}
|
||||
|
||||
// startSession handles POST /api/replay/start.
|
||||
// Creates a new replay session for the specified time range.
|
||||
func (h *ReplayHandler) startSession(w http.ResponseWriter, r *http.Request) {
|
||||
var req startSessionRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid request body", http.StatusBadRequest)
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body: " + err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
fromMS, err := parseISO8601(req.FromISO8601)
|
||||
if err != nil {
|
||||
http.Error(w, "invalid from_iso8601: "+err.Error(), http.StatusBadRequest)
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid from_iso8601: " + err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -141,13 +210,13 @@ func (h *ReplayHandler) startSession(w http.ResponseWriter, r *http.Request) {
|
|||
if req.ToISO8601 != "" {
|
||||
toMS, err = parseISO8601(req.ToISO8601)
|
||||
if err != nil {
|
||||
http.Error(w, "invalid to_iso8601: "+err.Error(), http.StatusBadRequest)
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid to_iso8601: " + err.Error()})
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if toMS < fromMS {
|
||||
http.Error(w, "to_iso8601 must be after from_iso8601", http.StatusBadRequest)
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "to_iso8601 must be after from_iso8601"})
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -156,7 +225,7 @@ func (h *ReplayHandler) startSession(w http.ResponseWriter, r *http.Request) {
|
|||
speed = 1
|
||||
}
|
||||
if speed != 1 && speed != 2 && speed != 5 {
|
||||
http.Error(w, "speed must be 1, 2, or 5", http.StatusBadRequest)
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "speed must be 1, 2, or 5"})
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -179,7 +248,7 @@ func (h *ReplayHandler) startSession(w http.ResponseWriter, r *http.Request) {
|
|||
log.Printf("[INFO] Replay session started: %s (from %d to %d, speed %dx)",
|
||||
session.ID, fromMS, toMS, speed)
|
||||
|
||||
writeJSON(w, map[string]interface{}{
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{
|
||||
"session_id": session.ID,
|
||||
"from_ms": fromMS,
|
||||
"to_ms": toMS,
|
||||
|
|
@ -188,14 +257,18 @@ func (h *ReplayHandler) startSession(w http.ResponseWriter, r *http.Request) {
|
|||
})
|
||||
}
|
||||
|
||||
// stopSessionRequest represents the request body for POST /api/replay/stop.
|
||||
type stopSessionRequest struct {
|
||||
// SessionID is the ID of the session to stop.
|
||||
SessionID string `json:"session_id"`
|
||||
}
|
||||
|
||||
// stopSession handles POST /api/replay/stop.
|
||||
// Stops the specified replay session and deletes it.
|
||||
func (h *ReplayHandler) stopSession(w http.ResponseWriter, r *http.Request) {
|
||||
var req stopSessionRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid request body", http.StatusBadRequest)
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body: " + err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -204,28 +277,35 @@ func (h *ReplayHandler) stopSession(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
session, exists := h.sessions[req.SessionID]
|
||||
if !exists {
|
||||
http.Error(w, "session not found", http.StatusNotFound)
|
||||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "session not found"})
|
||||
return
|
||||
}
|
||||
|
||||
session.State = "stopped"
|
||||
delete(h.sessions, req.SessionID)
|
||||
|
||||
writeJSON(w, map[string]interface{}{
|
||||
"status": "stopped",
|
||||
log.Printf("[INFO] Replay session stopped: %s", req.SessionID)
|
||||
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{
|
||||
"status": "stopped",
|
||||
"session": req.SessionID,
|
||||
})
|
||||
}
|
||||
|
||||
// seekRequest represents the request body for POST /api/replay/seek.
|
||||
type seekRequest struct {
|
||||
SessionID string `json:"session_id"`
|
||||
// SessionID is the ID of the session to seek within.
|
||||
SessionID string `json:"session_id"`
|
||||
// TimestampISO8601 is the target timestamp in ISO8601 format.
|
||||
TimestampISO8601 string `json:"timestamp_iso8601"`
|
||||
}
|
||||
|
||||
// seek handles POST /api/replay/seek.
|
||||
// Seeks to the specified timestamp within the session.
|
||||
func (h *ReplayHandler) seek(w http.ResponseWriter, r *http.Request) {
|
||||
var req seekRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid request body", http.StatusBadRequest)
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body: " + err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -234,18 +314,18 @@ func (h *ReplayHandler) seek(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
session, exists := h.sessions[req.SessionID]
|
||||
if !exists {
|
||||
http.Error(w, "session not found", http.StatusNotFound)
|
||||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "session not found"})
|
||||
return
|
||||
}
|
||||
|
||||
targetMS, err := parseISO8601(req.TimestampISO8601)
|
||||
if err != nil {
|
||||
http.Error(w, "invalid timestamp: "+err.Error(), http.StatusBadRequest)
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid timestamp: " + err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
if targetMS < session.FromMS || targetMS > session.ToMS {
|
||||
http.Error(w, "timestamp outside session range", http.StatusBadRequest)
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "timestamp outside session range"})
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -263,26 +343,37 @@ func (h *ReplayHandler) seek(w http.ResponseWriter, r *http.Request) {
|
|||
return true
|
||||
})
|
||||
|
||||
writeJSON(w, map[string]interface{}{
|
||||
log.Printf("[INFO] Replay session seeked: %s to %d", req.SessionID, targetMS)
|
||||
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{
|
||||
"status": "seeked",
|
||||
"current_ms": targetMS,
|
||||
"frame_found": len(frameData) > 0,
|
||||
})
|
||||
}
|
||||
|
||||
// tuneRequest represents the request body for POST /api/replay/tune.
|
||||
type tuneRequest struct {
|
||||
SessionID string `json:"session_id"`
|
||||
DeltaRMSThreshold *float64 `json:"delta_rms_threshold,omitempty"`
|
||||
TauS *float64 `json:"tau_s,omitempty"`
|
||||
FresnelDecay *float64 `json:"fresnel_decay,omitempty"`
|
||||
Subcarriers *int `json:"n_subcarriers,omitempty"`
|
||||
BreathingSensitivity *float64 `json:"breathing_sensitivity,omitempty"`
|
||||
// SessionID is the ID of the session to tune.
|
||||
SessionID string `json:"session_id"`
|
||||
// DeltaRMSThreshold is the motion detection threshold (0.001-1.0).
|
||||
DeltaRMSThreshold *float64 `json:"delta_rms_threshold,omitempty"`
|
||||
// TauS is the EMA baseline time constant in seconds (1-600).
|
||||
TauS *float64 `json:"tau_s,omitempty"`
|
||||
// FresnelDecay is the Fresnel zone weight decay rate (1.0-4.0).
|
||||
FresnelDecay *float64 `json:"fresnel_decay,omitempty"`
|
||||
// Subcarriers is the number of subcarriers for NBVI selection (8-47).
|
||||
Subcarriers *int `json:"n_subcarriers,omitempty"`
|
||||
// BreathingSensitivity is the breathing detection threshold in radians RMS (0.001-0.1).
|
||||
BreathingSensitivity *float64 `json:"breathing_sensitivity,omitempty"`
|
||||
}
|
||||
|
||||
// tune handles POST /api/replay/tune.
|
||||
// Updates pipeline parameters for the replay session.
|
||||
func (h *ReplayHandler) tune(w http.ResponseWriter, r *http.Request) {
|
||||
var req tuneRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid request body", http.StatusBadRequest)
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body: " + err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -291,7 +382,7 @@ func (h *ReplayHandler) tune(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
session, exists := h.sessions[req.SessionID]
|
||||
if !exists {
|
||||
http.Error(w, "session not found", http.StatusNotFound)
|
||||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "session not found"})
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -315,7 +406,7 @@ func (h *ReplayHandler) tune(w http.ResponseWriter, r *http.Request) {
|
|||
|
||||
log.Printf("[INFO] Replay session tuned: %s params=%+v", req.SessionID, params)
|
||||
|
||||
writeJSON(w, map[string]interface{}{
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{
|
||||
"status": "tuned",
|
||||
"params": params,
|
||||
"session": req.SessionID,
|
||||
|
|
|
|||
1137
mothership/internal/api/replay_test.go
Normal file
1137
mothership/internal/api/replay_test.go
Normal file
File diff suppressed because it is too large
Load diff
Loading…
Add table
Reference in a new issue