mk6i-open-oscar-server/server/webapi/handlers/events.go

package handlers

import (
	"context"
	"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
}

// FetchEventsData contains the events and metadata.
type FetchEventsData struct {
	Events          []types.Event `json:"events" xml:"events>event"`
	LastSeqNum      uint64        `json:"lastSeqNum" xml:"lastSeqNum"`
	TimeToNextFetch int           `json:"timeToNextFetch" xml:"timeToNextFetch"`
	FetchBaseURL    string        `json:"fetchBaseURL" xml:"fetchBaseURL"`
}

// 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, r, 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 := BaseResponse{}
	resp.Response.StatusCode = 200
	resp.Response.StatusText = "OK"
	resp.Response.Data = &FetchEventsData{
		Events:          events,
		LastSeqNum:      newLastSeqNum,
		TimeToNextFetch: session.TimeToNextFetch,
		// Include fetchBaseURL with updated sequence number for next request
		FetchBaseURL: fmt.Sprintf("%s/aim/fetchEvents?aimsid=%s&seqNum=%d",
			baseURLFromRequest(r), aimsid, newLastSeqNum),
	}

	// AMF3 clients (e.g. Gromit) take the events reshaped: timestamps as floats
	// and the source/dest user objects flattened. That is a payload difference,
	// not just an encoding one, so it stays here rather than in the encoder.
	format := strings.ToLower(r.URL.Query().Get("f"))
	if format == "amf" || format == "amf3" {
		amfResp := map[string]interface{}{
			"response": map[string]interface{}{
				"data": map[string]interface{}{
					"events":          ConvertEventsForAMF3(events),
					"lastSeqNum":      newLastSeqNum,
					"timeToNextFetch": session.TimeToNextFetch,
					"fetchBaseURL": fmt.Sprintf("%s/aim/fetchEvents?aimsid=%s&seqNum=%d",
						baseURLFromRequest(r), aimsid, newLastSeqNum),
				},
				"statusCode":       200,
				"statusText":       "OK",
				"statusDetailCode": 0,
			},
		}
		SendResponse(w, r, amfResp, h.Logger)
	} else {
		// Send response in requested format (JSON, JSONP, or XML)
		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, r *http.Request, statusCode int, message string) {
	SendError(w, r, statusCode, message)
}