package handlers import ( "context" "encoding/xml" "fmt" "log/slog" "net/http" "strconv" "strings" "time" "github.com/mk6i/open-oscar-server/server/webapi/types" "github.com/mk6i/open-oscar-server/state" ) // EventsHandler handles Web AIM API event fetching endpoints. type EventsHandler struct { SessionManager *state.WebAPISessionManager Logger *slog.Logger } // FetchEventsResponse represents the response for fetchEvents endpoint. type FetchEventsResponse struct { Response struct { StatusCode int `json:"statusCode"` StatusText string `json:"statusText"` Data FetchEventsData `json:"data"` } `json:"response"` } // FetchEventsData contains the events and metadata. type FetchEventsData struct { Events []types.Event `json:"events"` LastSeqNum uint64 `json:"lastSeqNum"` TimeToNextFetch int `json:"timeToNextFetch"` FetchBaseURL string `json:"fetchBaseURL"` } // FetchEventsXMLResponse represents the XML response for fetchEvents endpoint. type FetchEventsXMLResponse struct { XMLName xml.Name `xml:"response"` StatusCode int `xml:"statusCode"` StatusText string `xml:"statusText"` Data struct { Events []types.Event `xml:"events>event"` LastSeqNum uint64 `xml:"lastSeqNum"` TimeToNextFetch int `xml:"timeToNextFetch"` FetchBaseURL string `xml:"fetchBaseURL"` } `xml:"data"` } // FetchEvents handles GET /aim/fetchEvents requests with long-polling support. func (h *EventsHandler) FetchEvents(w http.ResponseWriter, r *http.Request, session *state.WebAPISession) { ctx := r.Context() aimsid := session.AimSID // Get sequence number parameter var lastSeqNum uint64 if seqStr := r.URL.Query().Get("seqNum"); seqStr != "" { if val, err := strconv.ParseUint(seqStr, 10, 64); err == nil { lastSeqNum = val } } // Timeout is in milliseconds (per Web API spec and client behavior). timeout := time.Duration(session.FetchTimeout) * time.Millisecond if timeoutStr := r.URL.Query().Get("timeout"); timeoutStr != "" { if val, err := strconv.Atoi(timeoutStr); err == nil && val > 0 { timeout = time.Duration(val) * time.Millisecond } } // Limit maximum timeout to 60 seconds if timeout > 60*time.Second { timeout = 60 * time.Second } // Create a context with timeout for the fetch operation fetchCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() // Fetch events from the queue (will block until events available or timeout) events, err := session.EventQueue.Fetch(fetchCtx, lastSeqNum, timeout) if err != nil { if err == context.DeadlineExceeded { // Timeout is normal - return empty events array events = []types.Event{} } else { h.Logger.ErrorContext(ctx, "failed to fetch events", "err", err.Error()) h.sendError(w, http.StatusInternalServerError, "failed to fetch events") return } } // Determine the last sequence number newLastSeqNum := lastSeqNum if len(events) > 0 { newLastSeqNum = events[len(events)-1].SeqNum } // Prepare response resp := FetchEventsResponse{} resp.Response.StatusCode = 200 resp.Response.StatusText = "OK" resp.Response.Data.Events = events resp.Response.Data.LastSeqNum = newLastSeqNum resp.Response.Data.TimeToNextFetch = session.TimeToNextFetch // Include fetchBaseURL with updated sequence number for next request resp.Response.Data.FetchBaseURL = fmt.Sprintf("http://%s/aim/fetchEvents?aimsid=%s&seqNum=%d", r.Host, aimsid, newLastSeqNum) // Check response format format := strings.ToLower(r.URL.Query().Get("f")) switch format { case "xml": // Send XML response xmlResp := FetchEventsXMLResponse{} xmlResp.StatusCode = 200 xmlResp.StatusText = "OK" xmlResp.Data.Events = events xmlResp.Data.LastSeqNum = newLastSeqNum xmlResp.Data.TimeToNextFetch = session.TimeToNextFetch xmlResp.Data.FetchBaseURL = fmt.Sprintf("http://%s/aim/fetchEvents?aimsid=%s&seqNum=%d", r.Host, aimsid, newLastSeqNum) w.Header().Set("Content-Type", "text/xml") _, _ = fmt.Fprint(w, ``) if err := xml.NewEncoder(w).Encode(xmlResp); err != nil { h.Logger.Error("failed to encode XML response", "error", err) } case "amf", "amf3": // For AMF3, build the response with fields in the correct order // The working implementation has: response { data {...}, statusCode, statusText, statusDetailCode } // Convert events to ensure timestamps are float64 for AMF3 convertedEvents := ConvertEventsForAMF3(events) amfResp := map[string]interface{}{ "response": map[string]interface{}{ // Data comes FIRST (Gromit processes this large object) "data": map[string]interface{}{ "events": convertedEvents, "lastSeqNum": newLastSeqNum, "timeToNextFetch": session.TimeToNextFetch, "fetchBaseURL": fmt.Sprintf("http://%s/aim/fetchEvents?aimsid=%s&seqNum=%d", r.Host, aimsid, newLastSeqNum), }, // Status fields come AFTER data "statusCode": 200, "statusText": "OK", "statusDetailCode": 0, }, } // Use SendResponse which will detect AMF format and encode properly SendResponse(w, r, amfResp, h.Logger) default: // Send JSON/JSONP response with standard structure SendResponse(w, r, resp, h.Logger) } if len(events) > 0 { h.Logger.DebugContext(ctx, "events fetched", "aimsid", aimsid, "count", len(events), "last_seq", newLastSeqNum, ) } } // sendError is a convenience method that wraps the common SendError function. func (h *EventsHandler) sendError(w http.ResponseWriter, statusCode int, message string) { SendError(w, statusCode, message) }