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

package types

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"time"
)

// EventType defines the type of WebAPI event.
type EventType string

const (
	EventTypeBuddyList    EventType = "buddylist"
	EventTypeConversation EventType = "conversation"
	EventTypeIM           EventType = "im"
	EventTypeOfflineIM    EventType = "offlineIM"
	EventTypePreference   EventType = "preference"
	EventTypePresence     EventType = "presence"
	EventTypeRateLimit    EventType = "rateLimit"
	EventTypeSentIM       EventType = "sentIM"
	EventTypeSessionEnded EventType = "sessionEnded"
	EventTypeStatus       EventType = "status"
	EventTypeTyping       EventType = "typing"
	EventTypePermitDeny   EventType = "permitDeny"
)

// Event represents an event to be delivered to a web client.
type Event struct {
	Type      EventType   `json:"type"`
	SeqNum    uint64      `json:"seqNum"`
	Timestamp int64       `json:"timestamp"`
	Data      interface{} `json:"eventData"`
}

// PresenceEvent represents a presence change event.
// Friendly repeats the viewer's alias for the user. The client's merge deletes any
// alias it already holds, so a presence update that omits it silently renames the
// buddy back to their screen name. See UserInfo.
type PresenceEvent struct {
	AimID      string `json:"aimId"`
	Friendly   string `json:"friendly,omitempty"`
	State      string `json:"state"` // "online", "offline", "away", "idle"
	StatusMsg  string `json:"statusMsg,omitempty"`
	AwayMsg    string `json:"awayMsg,omitempty"`
	IdleTime   int    `json:"idleTime,omitempty"`   // Minutes idle
	OnlineTime int64  `json:"onlineTime,omitempty"` // Unix timestamp
	UserType   string `json:"userType"`             // "aim", "icq", "admin"
}

// IMEvent represents an instant message event.
type IMEvent struct {
	Source    UserInfo `json:"source"`
	Message   string   `json:"message"`
	MsgID     string   `json:"msgId,omitempty"`
	Timestamp float64  `json:"timestamp"` // float64 for AMF3 encoding
	AutoResp  bool     `json:"autoresponse,omitempty"`
}

// SentIMEvent represents a sent instant message event.
type SentIMEvent struct {
	Sender    UserInfo `json:"sender"` // Sender user info
	Dest      UserInfo `json:"dest"`   // Destination user info
	Message   string   `json:"message"`
	MsgID     string   `json:"msgId,omitempty"`
	Timestamp float64  `json:"timestamp"` // float64 for AMF3 encoding
	AutoResp  bool     `json:"autoResponse,omitempty"`
}

// UserInfo represents basic user information in events.
// AimID is the normalized screen name the client keys users by. DisplayID is the
// screen name as its owner formatted it. Friendly is the viewer's private alias for
// that user, and takes precedence over DisplayID when the client renders a name.
//
// The client merges every user map it receives onto the single user object it holds
// per aimId, and that merge deletes friendly before applying the map. An alias
// therefore has to be repeated on every user map, or it is lost.
type UserInfo struct {
	AimID      string  `json:"aimId"`
	DisplayID  string  `json:"displayId,omitempty"`
	Friendly   string  `json:"friendly,omitempty"`
	UserType   string  `json:"userType,omitempty"`
	State      string  `json:"state,omitempty"`
	OnlineTime float64 `json:"onlineTime,omitempty"` // float64 for AMF3 encoding
}

// TypingEvent represents a typing notification event.
type TypingEvent struct {
	AimID        string `json:"aimId"`
	TypingStatus string `json:"typingStatus"`
}

// EventQueue manages a queue of events for a WebAPI session.
type EventQueue struct {
	events   []Event
	seqNum   uint64
	maxSize  int
	mu       sync.RWMutex
	waitChan chan struct{}
	closed   bool
	closedMu sync.RWMutex
}

// NewEventQueue creates a new event queue with the specified maximum size.
func NewEventQueue(maxSize int) *EventQueue {
	return &EventQueue{
		events:   make([]Event, 0),
		maxSize:  maxSize,
		waitChan: make(chan struct{}, 1),
	}
}

// Push adds an event to the queue.
func (q *EventQueue) Push(eventType EventType, data interface{}) {
	q.closedMu.RLock()
	if q.closed {
		q.closedMu.RUnlock()
		return
	}
	q.closedMu.RUnlock()

	q.mu.Lock()
	defer q.mu.Unlock()

	// Increment sequence number atomically
	seqNum := atomic.AddUint64(&q.seqNum, 1)

	event := Event{
		Type:      eventType,
		SeqNum:    seqNum,
		Timestamp: time.Now().Unix(),
		Data:      data,
	}

	// Add event to queue
	q.events = append(q.events, event)

	// If queue exceeds max size, remove oldest events
	if len(q.events) > q.maxSize {
		// Keep only the most recent maxSize events
		q.events = q.events[len(q.events)-q.maxSize:]
	}

	// Signal any waiting fetchers
	select {
	case q.waitChan <- struct{}{}:
	default:
		// Channel already has a signal
	}
}

// Fetch retrieves events from the queue, optionally waiting for new events.
func (q *EventQueue) Fetch(ctx context.Context, lastSeqNum uint64, timeout time.Duration) ([]Event, error) {
	q.closedMu.RLock()
	if q.closed {
		q.closedMu.RUnlock()
		return nil, errors.New("event queue is closed")
	}
	q.closedMu.RUnlock()

	// First, check if we have any events newer than lastSeqNum
	q.mu.RLock()
	events := q.getEventsAfter(lastSeqNum)
	q.mu.RUnlock()

	if len(events) > 0 {
		return events, nil
	}

	// No events available, wait for new ones or timeout
	timeoutChan := time.After(timeout)

	for {
		select {
		case <-q.waitChan:
			// New events may be available
			q.mu.RLock()
			events = q.getEventsAfter(lastSeqNum)
			q.mu.RUnlock()

			if len(events) > 0 {
				return events, nil
			}
			// False alarm, keep waiting

		case <-timeoutChan:
			// Timeout reached, return empty array
			return []Event{}, nil

		case <-ctx.Done():
			// Context cancelled
			return nil, ctx.Err()
		}
	}
}

// getEventsAfter returns all events with sequence number greater than the specified value.
// Must be called with at least a read lock held.
func (q *EventQueue) getEventsAfter(seqNum uint64) []Event {
	var result []Event

	for _, event := range q.events {
		if event.SeqNum > seqNum {
			result = append(result, event)
		}
	}

	return result
}

// GetAllEvents returns all events in the queue (for debugging).
func (q *EventQueue) GetAllEvents() []Event {
	q.mu.RLock()
	defer q.mu.RUnlock()

	result := make([]Event, len(q.events))
	copy(result, q.events)
	return result
}

// Close closes the event queue, unblocking any waiting fetchers.
func (q *EventQueue) Close() {
	q.closedMu.Lock()
	defer q.closedMu.Unlock()

	if q.closed {
		return
	}

	q.closed = true

	// Send multiple signals to unblock all potential waiters
notifyWaiters:
	for i := 0; i < 10; i++ {
		select {
		case q.waitChan <- struct{}{}:
		default:
			break notifyWaiters
		}
	}
}