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 (
	// Event types that can be subscribed to
	EventTypeBuddyList    EventType = "buddylist"
	EventTypePresence     EventType = "presence"
	EventTypeIM           EventType = "im"
	EventTypeSentIM       EventType = "sentIM"
	EventTypeTyping       EventType = "typing"
	EventTypeStatus       EventType = "status"
	EventTypeOfflineIM    EventType = "offlineIM"
	EventTypeSessionEnded EventType = "sessionEnded"
	EventTypeRateLimit    EventType = "rateLimit"
)

// 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:"data"`
}

// PresenceEvent represents a presence change event.
type PresenceEvent struct {
	AimID      string `json:"aimId"`
	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 {
	From      string  `json:"from"`
	Message   string  `json:"message"`
	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"`
	Timestamp float64  `json:"timestamp"` // float64 for AMF3 encoding
	AutoResp  bool     `json:"autoResponse,omitempty"`
}

// UserInfo represents basic user information in events.
type UserInfo struct {
	AimID      string  `json:"aimId"`
	DisplayID  string  `json:"displayId,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 {
	From   string `json:"from"`
	Typing bool   `json:"typing"`
}

// BuddyListEvent represents a buddy list change event.
type BuddyListEvent struct {
	Action string      `json:"action"` // "add", "remove", "update"
	Buddy  interface{} `json:"buddy"`
	Group  string      `json:"group,omitempty"`
}

// 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
}

// Clear removes all events from the queue.
func (q *EventQueue) Clear() {
	q.mu.Lock()
	defer q.mu.Unlock()

	q.events = make([]Event, 0)
}

// Size returns the current number of events in the queue.
func (q *EventQueue) Size() int {
	q.mu.RLock()
	defer q.mu.RUnlock()

	return len(q.events)
}

// 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
		}
	}
}

// IsClosed returns whether the queue is closed.
func (q *EventQueue) IsClosed() bool {
	q.closedMu.RLock()
	defer q.closedMu.RUnlock()
	return q.closed
}