package service

import (
	"sync"

	"browser.local/platform/domain"
	"browser.local/platform/validator"
)

const defaultRunControlEventRetrySeconds = 5

type RunControlEventSubscription struct {
	Initial *domain.RunControlEvent
	Events  <-chan domain.RunControlEvent
	Cancel  func()
}

func (svc *CoreService) SubscribeRunControlEvents(request domain.RunControlStreamRequest) (RunControlEventSubscription, error) {
	request = domain.CopyRunControlStreamRequest(request)
	if err := validator.ValidateRunControlStreamRequest(request); err != nil {
		return RunControlEventSubscription{}, err
	}
	if err := svc.validateRunSession(request.RunEndpointID, request.SessionToken); err != nil {
		return RunControlEventSubscription{}, err
	}

	updates := make(chan domain.RunControlEvent, 8)
	var initial *domain.RunControlEvent
	svc.controlStreamMu.Lock()
	if latest, exists := svc.controlStreamEvents[request.RunEndpointID]; exists && latest.Sequence > request.LastEventSeq {
		copy := domain.CopyRunControlEvent(latest)
		initial = &copy
	}
	svc.controlStreamWaiters[request.RunEndpointID] = append(svc.controlStreamWaiters[request.RunEndpointID], updates)
	svc.controlStreamMu.Unlock()

	var once sync.Once
	cancel := func() {
		once.Do(func() {
			svc.controlStreamMu.Lock()
			waiters := svc.controlStreamWaiters[request.RunEndpointID]
			for index, candidate := range waiters {
				if candidate == updates {
					waiters = append(waiters[:index], waiters[index+1:]...)
					break
				}
			}
			if len(waiters) == 0 {
				delete(svc.controlStreamWaiters, request.RunEndpointID)
			} else {
				svc.controlStreamWaiters[request.RunEndpointID] = waiters
			}
			svc.controlStreamMu.Unlock()
			close(updates)
		})
	}
	return RunControlEventSubscription{Initial: initial, Events: updates, Cancel: cancel}, nil
}

func (svc *CoreService) publishRunControlEvent(runEndpointID string, eventType string) {
	if runEndpointID == "" || eventType == "" {
		return
	}
	svc.controlStreamMu.Lock()
	sequence := svc.controlStreamSeq[runEndpointID] + 1
	svc.controlStreamSeq[runEndpointID] = sequence
	event := domain.RunControlEvent{RunEndpointID: runEndpointID, Sequence: sequence, Type: eventType, ServerTime: svc.now(), RetrySeconds: defaultRunControlEventRetrySeconds}
	svc.controlStreamEvents[runEndpointID] = event
	waiters := append([]chan domain.RunControlEvent(nil), svc.controlStreamWaiters[runEndpointID]...)
	svc.controlStreamMu.Unlock()
	for _, waiter := range waiters {
		select {
		case waiter <- event:
		default:
		}
	}
}