package websocket

import (
	"bytes"
	"context"
	"errors"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	gorillawebsocket "github.com/gorilla/websocket"
	"github.com/pingcap-inc/nexuscrm-api-host/hostruntime"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionstatev1"
	"github.com/rs/zerolog"
	"github.com/rs/zerolog/log"
)

type authorizingWebSocketStore struct {
	mu               sync.Mutex
	channels         map[string]map[string]bool
	destroyed        map[string]bool
	subscribeStarted chan struct{}
	subscribeRelease <-chan struct{}
	subscribeOnce    sync.Once
}

func newAuthorizingWebSocketStore() *authorizingWebSocketStore {
	return &authorizingWebSocketStore{
		channels:  make(map[string]map[string]bool),
		destroyed: make(map[string]bool),
	}
}

func (store *authorizingWebSocketStore) CreateSession() string {
	store.mu.Lock()
	defer store.mu.Unlock()
	const clientID = "client-1"
	store.channels[clientID] = make(map[string]bool)
	return clientID
}

func (store *authorizingWebSocketStore) DestroySession(clientID string) {
	store.mu.Lock()
	defer store.mu.Unlock()
	delete(store.channels, clientID)
	store.destroyed[clientID] = true
}

func (store *authorizingWebSocketStore) Subscribe(
	clientID string,
	channel string,
) bool {
	if store.subscribeStarted != nil {
		store.subscribeOnce.Do(func() {
			close(store.subscribeStarted)
		})
	}
	if store.subscribeRelease != nil {
		<-store.subscribeRelease
	}
	store.mu.Lock()
	defer store.mu.Unlock()
	if _, ok := store.channels[clientID]; !ok {
		return false
	}
	store.channels[clientID][channel] = true
	return true
}

func (store *authorizingWebSocketStore) wasDestroyed(clientID string) bool {
	store.mu.Lock()
	defer store.mu.Unlock()
	return store.destroyed[clientID]
}

func (store *authorizingWebSocketStore) Unsubscribe(
	clientID string,
	channel string,
) bool {
	store.mu.Lock()
	defer store.mu.Unlock()
	if _, ok := store.channels[clientID]; !ok {
		return false
	}
	delete(store.channels[clientID], channel)
	return true
}

func (store *authorizingWebSocketStore) GetSubscribedChannels(
	clientID string,
) []string {
	store.mu.Lock()
	defer store.mu.Unlock()
	var result []string
	for channel := range store.channels[clientID] {
		result = append(result, channel)
	}
	return result
}

func (store *authorizingWebSocketStore) subscribed(
	clientID string,
	channel string,
) bool {
	store.mu.Lock()
	defer store.mu.Unlock()
	return store.channels[clientID][channel]
}

type authorizingWebSocketLease struct {
	mu       sync.Mutex
	results  []hostruntime.AuthorizingPermissionState
	errors   []error
	requests []hostruntime.WebSocketAuthorizationRequest
	wait     bool
}

func (lease *authorizingWebSocketLease) ReauthorizeWebSocket(
	ctx context.Context,
	request hostruntime.WebSocketAuthorizationRequest,
) (hostruntime.AuthorizingPermissionState, error) {
	lease.mu.Lock()
	index := len(lease.requests)
	lease.requests = append(lease.requests, request)
	wait := lease.wait
	var result hostruntime.AuthorizingPermissionState
	if index < len(lease.results) {
		result = lease.results[index]
	}
	var err error
	if index < len(lease.errors) {
		err = lease.errors[index]
	}
	lease.mu.Unlock()
	if wait {
		<-ctx.Done()
		return hostruntime.AuthorizingPermissionState{}, ctx.Err()
	}
	return result, err
}

func (lease *authorizingWebSocketLease) capturedRequests() []hostruntime.WebSocketAuthorizationRequest {
	lease.mu.Lock()
	defer lease.mu.Unlock()
	return append(
		[]hostruntime.WebSocketAuthorizationRequest(nil),
		lease.requests...,
	)
}

type authorizingWebSocketFactory struct {
	lease   hostruntime.WebSocketAuthorizationLease
	err     error
	bearer  string
	initial hostruntime.AuthorizingPermissionState
}

type cancellationIgnoringWebSocketLease struct {
	started chan struct{}
	release <-chan struct{}
	state   hostruntime.AuthorizingPermissionState
}

func (lease *cancellationIgnoringWebSocketLease) ReauthorizeWebSocket(
	context.Context,
	hostruntime.WebSocketAuthorizationRequest,
) (hostruntime.AuthorizingPermissionState, error) {
	close(lease.started)
	<-lease.release
	return lease.state, nil
}

type cancellationIgnoringWebSocketFactory struct {
	started chan struct{}
	release <-chan struct{}
	lease   hostruntime.WebSocketAuthorizationLease
}

func (factory *cancellationIgnoringWebSocketFactory) OpenWebSocketAuthorization(
	context.Context,
	string,
	hostruntime.AuthorizingPermissionState,
) (hostruntime.WebSocketAuthorizationLease, error) {
	close(factory.started)
	<-factory.release
	return factory.lease, nil
}

func (factory *authorizingWebSocketFactory) OpenWebSocketAuthorization(
	_ context.Context,
	bearerToken string,
	initial hostruntime.AuthorizingPermissionState,
) (hostruntime.WebSocketAuthorizationLease, error) {
	factory.bearer = bearerToken
	factory.initial = initial
	return factory.lease, factory.err
}

func TestNewAuthorizingWebSocketHandlerRejectsIncompleteComposition(
	t *testing.T,
) {
	store := newAuthorizingWebSocketStore()
	factory := &authorizingWebSocketFactory{
		lease: &authorizingWebSocketLease{},
	}
	tests := []struct {
		name    string
		options AuthorizingWebSocketHandlerOptions
	}{
		{
			name: "session manager",
			options: AuthorizingWebSocketHandlerOptions{
				AuthorizationFactory:   factory,
				ReauthorizationTimeout: time.Second,
				WriteTimeout:           time.Second,
			},
		},
		{
			name: "authorization factory",
			options: AuthorizingWebSocketHandlerOptions{
				SessionManager:         store,
				ReauthorizationTimeout: time.Second,
				WriteTimeout:           time.Second,
			},
		},
		{
			name: "positive timeout",
			options: AuthorizingWebSocketHandlerOptions{
				SessionManager:       store,
				AuthorizationFactory: factory,
				WriteTimeout:         time.Second,
			},
		},
		{
			name: "positive write timeout",
			options: AuthorizingWebSocketHandlerOptions{
				SessionManager:         store,
				AuthorizationFactory:   factory,
				ReauthorizationTimeout: time.Second,
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			handler, err := NewAuthorizingWebSocketHandler(test.options)
			if err == nil || handler != nil {
				t.Fatalf(
					"constructor without %s = %#v, %v",
					test.name,
					handler,
					err,
				)
			}
		})
	}
}

func TestAuthorizingWebSocketReauthorizesSubscribeAndDelivery(t *testing.T) {
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	lease := &authorizingWebSocketLease{
		results: []hostruntime.AuthorizingPermissionState{
			authorizingWebSocketState(t, 1, 1, "user-1"),
			authorizingWebSocketState(t, 1, 2, "user-1"),
		},
	}
	factory := &authorizingWebSocketFactory{lease: lease}
	store := newAuthorizingWebSocketStore()
	handler, err := NewAuthorizingWebSocketHandler(
		AuthorizingWebSocketHandlerOptions{
			SessionManager:         store,
			AuthorizationFactory:   factory,
			ReauthorizationTimeout: time.Second,
			WriteTimeout:           time.Second,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	server := authorizingWebSocketServer(t, handler, initial)
	connection := dialAuthorizingWebSocket(t, server.URL, "strong-bearer")
	defer connection.Close()

	readAuthorizingWebSocketJSON(t, connection)
	if err := connection.WriteJSON(map[string]any{
		"channel":      "/meta/subscribe",
		"subscription": "/events/current",
	}); err != nil {
		t.Fatal(err)
	}
	readAuthorizingWebSocketJSON(t, connection)
	if !store.subscribed("client-1", "/events/current") {
		t.Fatal("subscription mutated without reaching the session store")
	}

	handler.Broadcast("/events/current", map[string]any{"current": true})
	delivery := readAuthorizingWebSocketJSON(t, connection)
	if delivery["channel"] != "/events/current" {
		t.Fatalf("delivery = %#v", delivery)
	}

	requests := lease.capturedRequests()
	if len(requests) != 2 ||
		requests[0].Operation != hostruntime.WebSocketAuthorizeSubscribe ||
		requests[0].Channel != "/events/current" ||
		requests[1].Operation != hostruntime.WebSocketAuthorizeDelivery ||
		requests[1].Channel != "/events/current" {
		t.Fatalf("reauthorization requests = %#v", requests)
	}
	if factory.bearer != "strong-bearer" ||
		factory.initial.Snapshot.Version().Revision() != 1 {
		t.Fatalf(
			"factory bearer/state = %q/%#v",
			factory.bearer,
			factory.initial,
		)
	}
}

func TestAuthorizingWebSocketClosesBeforeUnauthorizedMutation(t *testing.T) {
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	tests := []struct {
		name     string
		lease    *authorizingWebSocketLease
		wantCode int
	}{
		{
			name: "credential rejected",
			lease: &authorizingWebSocketLease{
				errors: []error{
					hostruntime.ErrAuthorizingCredentialRejected,
				},
			},
			wantCode: gorillawebsocket.ClosePolicyViolation,
		},
		{
			name: "authority unavailable",
			lease: &authorizingWebSocketLease{
				errors: []error{errors.New("authority unavailable")},
			},
			wantCode: gorillawebsocket.CloseInternalServerErr,
		},
		{
			name: "lineage changed",
			lease: &authorizingWebSocketLease{
				results: []hostruntime.AuthorizingPermissionState{
					authorizingWebSocketState(t, 2, 1, "user-1"),
				},
			},
			wantCode: gorillawebsocket.CloseInternalServerErr,
		},
		{
			name: "subject changed",
			lease: &authorizingWebSocketLease{
				results: []hostruntime.AuthorizingPermissionState{
					authorizingWebSocketState(t, 1, 2, "user-2"),
				},
			},
			wantCode: gorillawebsocket.CloseInternalServerErr,
		},
		{
			name: "same version changed system permissions",
			lease: &authorizingWebSocketLease{
				results: []hostruntime.AuthorizingPermissionState{
					authorizingWebSocketStateWithPermissions(
						t,
						1,
						1,
						"user-1",
						[]string{"ApiEnabled", "ModifyAllData"},
						nil,
					),
				},
			},
			wantCode: gorillawebsocket.CloseInternalServerErr,
		},
		{
			name: "same version changed custom permissions",
			lease: &authorizingWebSocketLease{
				results: []hostruntime.AuthorizingPermissionState{
					authorizingWebSocketStateWithPermissions(
						t,
						1,
						1,
						"user-1",
						[]string{"ApiEnabled"},
						[]string{"CanStream"},
					),
				},
			},
			wantCode: gorillawebsocket.CloseInternalServerErr,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			store := newAuthorizingWebSocketStore()
			handler, err := NewAuthorizingWebSocketHandler(
				AuthorizingWebSocketHandlerOptions{
					SessionManager: store,
					AuthorizationFactory: &authorizingWebSocketFactory{
						lease: test.lease,
					},
					ReauthorizationTimeout: time.Second,
					WriteTimeout:           time.Second,
				},
			)
			if err != nil {
				t.Fatal(err)
			}
			server := authorizingWebSocketServer(t, handler, initial)
			connection := dialAuthorizingWebSocket(
				t,
				server.URL,
				"strong-bearer",
			)
			defer connection.Close()
			readAuthorizingWebSocketJSON(t, connection)

			if err := connection.WriteJSON(map[string]any{
				"channel":      "/meta/subscribe",
				"subscription": "/events/forbidden",
			}); err != nil {
				t.Fatal(err)
			}
			_, _, err = connection.ReadMessage()
			var closeError *gorillawebsocket.CloseError
			if !errors.As(err, &closeError) ||
				closeError.Code != test.wantCode {
				t.Fatalf(
					"close error = %#v, %v; want code %d",
					closeError,
					err,
					test.wantCode,
				)
			}
			if store.subscribed("client-1", "/events/forbidden") {
				t.Fatal("authorization failure allowed subscription mutation")
			}
		})
	}
}

func TestAuthorizingWebSocketClosesBeforeUnauthorizedDelivery(t *testing.T) {
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	lease := &authorizingWebSocketLease{
		results: []hostruntime.AuthorizingPermissionState{
			authorizingWebSocketState(t, 1, 2, "user-1"),
		},
		errors: []error{
			nil,
			errors.New("delivery authority unavailable"),
		},
	}
	store := newAuthorizingWebSocketStore()
	handler, err := NewAuthorizingWebSocketHandler(
		AuthorizingWebSocketHandlerOptions{
			SessionManager: store,
			AuthorizationFactory: &authorizingWebSocketFactory{
				lease: lease,
			},
			ReauthorizationTimeout: time.Second,
			WriteTimeout:           time.Second,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	server := authorizingWebSocketServer(t, handler, initial)
	connection := dialAuthorizingWebSocket(t, server.URL, "strong-bearer")
	defer connection.Close()
	readAuthorizingWebSocketJSON(t, connection)

	if err := connection.WriteJSON(map[string]any{
		"channel":      "/meta/subscribe",
		"subscription": "/events/current",
	}); err != nil {
		t.Fatal(err)
	}
	readAuthorizingWebSocketJSON(t, connection)

	handler.Broadcast("/events/current", map[string]any{"must_not_arrive": true})
	_, _, err = connection.ReadMessage()
	var closeError *gorillawebsocket.CloseError
	if !errors.As(err, &closeError) ||
		closeError.Code != gorillawebsocket.CloseInternalServerErr {
		t.Fatalf("delivery close error = %#v, %v", closeError, err)
	}
	requests := lease.capturedRequests()
	if len(requests) != 2 ||
		requests[1].Operation != hostruntime.WebSocketAuthorizeDelivery {
		t.Fatalf("delivery reauthorization requests = %#v", requests)
	}
}

func TestAuthorizingWebSocketBoundsReauthorization(t *testing.T) {
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	store := newAuthorizingWebSocketStore()
	handler, err := NewAuthorizingWebSocketHandler(
		AuthorizingWebSocketHandlerOptions{
			SessionManager: store,
			AuthorizationFactory: &authorizingWebSocketFactory{
				lease: &authorizingWebSocketLease{wait: true},
			},
			ReauthorizationTimeout: 20 * time.Millisecond,
			WriteTimeout:           time.Second,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	server := authorizingWebSocketServer(t, handler, initial)
	connection := dialAuthorizingWebSocket(t, server.URL, "strong-bearer")
	defer connection.Close()
	readAuthorizingWebSocketJSON(t, connection)

	started := time.Now()
	if err := connection.WriteJSON(map[string]any{
		"channel":      "/meta/subscribe",
		"subscription": "/events/bounded",
	}); err != nil {
		t.Fatal(err)
	}
	_, _, err = connection.ReadMessage()
	var closeError *gorillawebsocket.CloseError
	if !errors.As(err, &closeError) ||
		closeError.Code != gorillawebsocket.CloseInternalServerErr {
		t.Fatalf("bounded close error = %#v, %v", closeError, err)
	}
	if elapsed := time.Since(started); elapsed > time.Second {
		t.Fatalf("bounded reauthorization took %s", elapsed)
	}
}

func TestAuthorizingWebSocketBoundsLeaseThatIgnoresCancellation(t *testing.T) {
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	started := make(chan struct{})
	release := make(chan struct{})
	lease := &cancellationIgnoringWebSocketLease{
		started: started,
		release: release,
		state:   authorizingWebSocketState(t, 1, 2, "user-1"),
	}
	client := &wsClient{
		authorization:          lease,
		authorizationTimeout:   20 * time.Millisecond,
		authorizedUserID:       "user-1",
		lastAuthorizationState: initial,
		authorizationGate:      make(chan struct{}, 1),
	}
	actionCalled := false
	startedAt := time.Now()
	err := client.Authorize(
		context.Background(),
		hostruntime.WebSocketAuthorizationRequest{
			Operation: hostruntime.WebSocketAuthorizeSubscribe,
			Channel:   "/events/bounded",
		},
		func() error {
			actionCalled = true
			return nil
		},
	)
	<-started
	close(release)
	if !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("cancellation-ignoring lease error = %v", err)
	}
	if actionCalled {
		t.Fatal("timed-out cancellation-ignoring lease executed its action")
	}
	if elapsed := time.Since(startedAt); elapsed > time.Second {
		t.Fatalf("cancellation-ignoring lease pinned caller for %s", elapsed)
	}
}

func TestAuthorizingWebSocketBoundsFactoryThatIgnoresCancellation(t *testing.T) {
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	started := make(chan struct{})
	release := make(chan struct{})
	handler, err := NewAuthorizingWebSocketHandler(
		AuthorizingWebSocketHandlerOptions{
			SessionManager: newAuthorizingWebSocketStore(),
			AuthorizationFactory: &cancellationIgnoringWebSocketFactory{
				started: started,
				release: release,
				lease:   &authorizingWebSocketLease{},
			},
			ReauthorizationTimeout: 20 * time.Millisecond,
			WriteTimeout:           time.Second,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	server := authorizingWebSocketServer(t, handler, initial)
	header := make(http.Header)
	header.Set("Authorization", "Bearer strong-bearer")
	startedAt := time.Now()
	connection, response, err := gorillawebsocket.DefaultDialer.Dial(
		"ws"+strings.TrimPrefix(server.URL, "http")+"/ws",
		header,
	)
	if connection != nil {
		connection.Close()
	}
	<-started
	close(release)
	if err == nil || response == nil ||
		response.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf(
			"dial with cancellation-ignoring factory = %#v, %#v, %v",
			connection,
			response,
			err,
		)
	}
	response.Body.Close()
	if elapsed := time.Since(startedAt); elapsed > time.Second {
		t.Fatalf("cancellation-ignoring factory pinned caller for %s", elapsed)
	}
}

func TestAuthorizingWebSocketCloseDoesNotWaitForAuthorizedAction(
	t *testing.T,
) {
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	subscribeStarted := make(chan struct{})
	subscribeRelease := make(chan struct{})
	defer func() {
		select {
		case <-subscribeRelease:
		default:
			close(subscribeRelease)
		}
	}()
	store := newAuthorizingWebSocketStore()
	store.subscribeStarted = subscribeStarted
	store.subscribeRelease = subscribeRelease
	handler, err := NewAuthorizingWebSocketHandler(
		AuthorizingWebSocketHandlerOptions{
			SessionManager: store,
			AuthorizationFactory: &authorizingWebSocketFactory{
				lease: &authorizingWebSocketLease{
					results: []hostruntime.AuthorizingPermissionState{
						authorizingWebSocketState(t, 1, 2, "user-1"),
					},
				},
			},
			ReauthorizationTimeout: time.Second,
			WriteTimeout:           100 * time.Millisecond,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	server := authorizingWebSocketServer(t, handler, initial)
	connection := dialAuthorizingWebSocket(t, server.URL, "strong-bearer")
	defer connection.Close()
	readAuthorizingWebSocketJSON(t, connection)
	if err := connection.WriteJSON(map[string]any{
		"channel":      "/meta/subscribe",
		"subscription": "/events/blocked",
	}); err != nil {
		t.Fatal(err)
	}
	select {
	case <-subscribeStarted:
	case <-time.After(time.Second):
		t.Fatal("authorized action did not start")
	}

	handler.mu.RLock()
	client := handler.clients["client-1"]
	handler.mu.RUnlock()
	if client == nil {
		t.Fatal("connected client is unavailable")
	}
	closeDone := make(chan struct{})
	go func() {
		handler.closeForAuthorizationFailure(
			"client-1",
			client,
			errors.New("authority unavailable"),
		)
		close(closeDone)
	}()

	_, _, err = connection.ReadMessage()
	var closeError *gorillawebsocket.CloseError
	if !errors.As(err, &closeError) ||
		closeError.Code != gorillawebsocket.CloseInternalServerErr {
		t.Fatalf("close error = %#v, %v", closeError, err)
	}
	select {
	case <-closeDone:
	case <-time.After(time.Second):
		t.Fatal("connection cleanup waited for the authorized action")
	}
	if store.wasDestroyed("client-1") {
		t.Fatal("connection cleanup destroyed the session before the admitted action completed")
	}
	close(subscribeRelease)
	deadline := time.Now().Add(time.Second)
	for !store.wasDestroyed("client-1") && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if !store.wasDestroyed("client-1") {
		t.Fatal("connection cleanup did not destroy the session after the admitted action completed")
	}
}

func TestAuthorizingWebSocketSerializesReauthorizationAndAction(t *testing.T) {
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	lease := &authorizingWebSocketLease{
		results: []hostruntime.AuthorizingPermissionState{
			authorizingWebSocketState(t, 1, 2, "user-1"),
		},
		errors: []error{
			nil,
			hostruntime.ErrAuthorizingCredentialRejected,
		},
	}
	client := &wsClient{
		authorization:          lease,
		authorizationTimeout:   time.Second,
		authorizedUserID:       "user-1",
		lastAuthorizationState: initial,
		authorizationGate:      make(chan struct{}, 1),
	}
	firstActionStarted := make(chan struct{})
	releaseFirstAction := make(chan struct{})
	firstDone := make(chan error, 1)
	go func() {
		firstDone <- client.Authorize(
			context.Background(),
			hostruntime.WebSocketAuthorizationRequest{
				Operation: hostruntime.WebSocketAuthorizeSubscribe,
				Channel:   "/events/current",
			},
			func() error {
				close(firstActionStarted)
				<-releaseFirstAction
				return nil
			},
		)
	}()
	<-firstActionStarted

	secondActionCalled := make(chan struct{}, 1)
	secondDone := make(chan error, 1)
	go func() {
		secondDone <- client.Authorize(
			context.Background(),
			hostruntime.WebSocketAuthorizationRequest{
				Operation: hostruntime.WebSocketAuthorizeDelivery,
				Channel:   "/events/current",
			},
			func() error {
				secondActionCalled <- struct{}{}
				return nil
			},
		)
	}()
	time.Sleep(20 * time.Millisecond)
	if requests := lease.capturedRequests(); len(requests) != 1 {
		t.Fatalf(
			"concurrent reauthorization entered before first action completed: %#v",
			requests,
		)
	}

	close(releaseFirstAction)
	if err := <-firstDone; err != nil {
		t.Fatalf("first authorization/action failed: %v", err)
	}
	if err := <-secondDone; !errors.Is(
		err,
		hostruntime.ErrAuthorizingCredentialRejected,
	) {
		t.Fatalf("second authorization error = %v", err)
	}
	select {
	case <-secondActionCalled:
		t.Fatal("rejected concurrent authorization executed its action")
	default:
	}

	thirdActionCalled := false
	err := client.Authorize(
		context.Background(),
		hostruntime.WebSocketAuthorizationRequest{
			Operation: hostruntime.WebSocketAuthorizeDelivery,
			Channel:   "/events/current",
		},
		func() error {
			thirdActionCalled = true
			return nil
		},
	)
	if err == nil || thirdActionCalled {
		t.Fatalf(
			"closed authorization accepted queued action: called=%t err=%v",
			thirdActionCalled,
			err,
		)
	}
	if requests := lease.capturedRequests(); len(requests) != 2 {
		t.Fatalf("closed authorization called lease again: %#v", requests)
	}
}

func TestAuthorizingWebSocketBoundsAuthorizationGateWait(t *testing.T) {
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	lease := &authorizingWebSocketLease{
		results: []hostruntime.AuthorizingPermissionState{
			authorizingWebSocketState(t, 1, 2, "user-1"),
		},
	}
	client := &wsClient{
		authorization:          lease,
		authorizationTimeout:   30 * time.Millisecond,
		authorizedUserID:       "user-1",
		lastAuthorizationState: initial,
		authorizationGate:      make(chan struct{}, 1),
	}
	firstActionStarted := make(chan struct{})
	releaseFirstAction := make(chan struct{})
	firstDone := make(chan error, 1)
	go func() {
		firstDone <- client.Authorize(
			context.Background(),
			hostruntime.WebSocketAuthorizationRequest{
				Operation: hostruntime.WebSocketAuthorizeSubscribe,
				Channel:   "/events/current",
			},
			func() error {
				close(firstActionStarted)
				<-releaseFirstAction
				return nil
			},
		)
	}()
	<-firstActionStarted

	started := time.Now()
	err := client.Authorize(
		context.Background(),
		hostruntime.WebSocketAuthorizationRequest{
			Operation: hostruntime.WebSocketAuthorizeDelivery,
			Channel:   "/events/current",
		},
		func() error {
			t.Fatal("timed-out gate wait executed action")
			return nil
		},
	)
	if !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("gate wait error = %v", err)
	}
	if elapsed := time.Since(started); elapsed > time.Second {
		t.Fatalf("authorization gate wait took %s", elapsed)
	}
	close(releaseFirstAction)
	if err := <-firstDone; err != nil {
		t.Fatalf("first authorization/action failed: %v", err)
	}
}

func TestAuthorizingWebSocketRejectsMissingInitialStateBeforeUpgrade(
	t *testing.T,
) {
	handler, err := NewAuthorizingWebSocketHandler(
		AuthorizingWebSocketHandlerOptions{
			SessionManager: newAuthorizingWebSocketStore(),
			AuthorizationFactory: &authorizingWebSocketFactory{
				lease: &authorizingWebSocketLease{},
			},
			ReauthorizationTimeout: time.Second,
			WriteTimeout:           time.Second,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	gin.SetMode(gin.TestMode)
	router := gin.New()
	router.GET("/ws", handler.HandleConnection)
	server := httptest.NewServer(router)
	t.Cleanup(server.Close)

	header := make(http.Header)
	header.Set("Authorization", "Bearer strong-bearer")
	connection, response, err := gorillawebsocket.DefaultDialer.Dial(
		"ws"+strings.TrimPrefix(server.URL, "http")+"/ws",
		header,
	)
	if connection != nil {
		connection.Close()
	}
	if err == nil || response == nil ||
		response.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf(
			"dial without initial state = %#v, %#v, %v",
			connection,
			response,
			err,
		)
	}
	response.Body.Close()
}

func TestAuthorizingWebSocketClassifiesLeaseOpenFailure(t *testing.T) {
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	tests := []struct {
		name       string
		err        error
		wantStatus int
	}{
		{
			name:       "credential rejected",
			err:        hostruntime.ErrAuthorizingCredentialRejected,
			wantStatus: http.StatusUnauthorized,
		},
		{
			name:       "authority unavailable",
			err:        errors.New("authority unavailable"),
			wantStatus: http.StatusServiceUnavailable,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			handler, err := NewAuthorizingWebSocketHandler(
				AuthorizingWebSocketHandlerOptions{
					SessionManager: newAuthorizingWebSocketStore(),
					AuthorizationFactory: &authorizingWebSocketFactory{
						err: test.err,
					},
					ReauthorizationTimeout: time.Second,
					WriteTimeout:           time.Second,
				},
			)
			if err != nil {
				t.Fatal(err)
			}
			server := authorizingWebSocketServer(t, handler, initial)
			header := make(http.Header)
			header.Set("Authorization", "Bearer strong-bearer")
			connection, response, err :=
				gorillawebsocket.DefaultDialer.Dial(
					"ws"+strings.TrimPrefix(server.URL, "http")+"/ws",
					header,
				)
			if connection != nil {
				connection.Close()
			}
			if err == nil || response == nil ||
				response.StatusCode != test.wantStatus {
				t.Fatalf(
					"lease open failure = %#v, %#v, %v; want %d",
					connection,
					response,
					err,
					test.wantStatus,
				)
			}
			response.Body.Close()
		})
	}
}

func TestAuthorizingWebSocketDoesNotLogLeaseOpenErrorDetails(t *testing.T) {
	const bearer = "strong-bearer-must-not-appear"
	initial := authorizingWebSocketState(t, 1, 1, "user-1")
	handler, err := NewAuthorizingWebSocketHandler(
		AuthorizingWebSocketHandlerOptions{
			SessionManager: newAuthorizingWebSocketStore(),
			AuthorizationFactory: &authorizingWebSocketFactory{
				err: errors.New("factory echoed " + bearer),
			},
			ReauthorizationTimeout: time.Second,
			WriteTimeout:           time.Second,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	var output bytes.Buffer
	previousLogger := log.Logger
	log.Logger = zerolog.New(&output)
	t.Cleanup(func() {
		log.Logger = previousLogger
	})

	server := authorizingWebSocketServer(t, handler, initial)
	header := make(http.Header)
	header.Set("Authorization", "Bearer "+bearer)
	connection, response, err := gorillawebsocket.DefaultDialer.Dial(
		"ws"+strings.TrimPrefix(server.URL, "http")+"/ws",
		header,
	)
	if connection != nil {
		connection.Close()
	}
	if err == nil || response == nil ||
		response.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf(
			"lease open failure = %#v, %#v, %v; want 503",
			connection,
			response,
			err,
		)
	}
	response.Body.Close()
	if strings.Contains(output.String(), bearer) {
		t.Fatalf("lease-open log exposed bearer: %s", output.String())
	}
}

func authorizingWebSocketServer(
	t *testing.T,
	handler *WebSocketHandler,
	initial hostruntime.AuthorizingPermissionState,
) *httptest.Server {
	t.Helper()
	gin.SetMode(gin.TestMode)
	router := gin.New()
	router.GET("/ws", func(c *gin.Context) {
		ctx, err := hostruntime.ContextWithAuthorizingPermissionState(
			c.Request.Context(),
			initial,
		)
		if err != nil {
			t.Error(err)
			c.Status(http.StatusInternalServerError)
			return
		}
		c.Request = c.Request.WithContext(ctx)
		handler.HandleConnection(c)
	})
	server := httptest.NewServer(router)
	t.Cleanup(server.Close)
	return server
}

func dialAuthorizingWebSocket(
	t *testing.T,
	serverURL string,
	bearer string,
) *gorillawebsocket.Conn {
	t.Helper()
	header := make(http.Header)
	header.Set("Authorization", "Bearer "+bearer)
	connection, response, err := gorillawebsocket.DefaultDialer.Dial(
		"ws"+strings.TrimPrefix(serverURL, "http")+"/ws",
		header,
	)
	if err != nil {
		if response != nil {
			response.Body.Close()
		}
		t.Fatal(err)
	}
	return connection
}

func readAuthorizingWebSocketJSON(
	t *testing.T,
	connection *gorillawebsocket.Conn,
) map[string]any {
	t.Helper()
	result := make(map[string]any)
	if err := connection.ReadJSON(&result); err != nil {
		t.Fatal(err)
	}
	return result
}

func authorizingWebSocketState(
	t *testing.T,
	lineageByte byte,
	revision uint64,
	userID string,
) hostruntime.AuthorizingPermissionState {
	t.Helper()
	return authorizingWebSocketStateWithPermissions(
		t,
		lineageByte,
		revision,
		userID,
		[]string{"ApiEnabled"},
		nil,
	)
}

func authorizingWebSocketStateWithPermissions(
	t *testing.T,
	lineageByte byte,
	revision uint64,
	userID string,
	systemPermissions []string,
	customPermissions []string,
) hostruntime.AuthorizingPermissionState {
	t.Helper()
	subject, err := permissionstatev1.NewSubject(userID, "profile-1")
	if err != nil {
		t.Fatal(err)
	}
	lineageBytes := make([]byte, permissionstatev1.LineageBytes)
	lineageBytes[0] = lineageByte
	lineage, err := permissionstatev1.LineageFromBytes(lineageBytes)
	if err != nil {
		t.Fatal(err)
	}
	version, err := permissionstatev1.NewVersion(lineage, revision)
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := permissionstatev1.NewEffectiveSnapshot(
		version,
		systemPermissions,
		customPermissions,
	)
	if err != nil {
		t.Fatal(err)
	}
	return hostruntime.AuthorizingPermissionState{
		Subject:  subject,
		Floor:    version,
		Snapshot: snapshot,
	}
}