requestloop.go

package jaws

// This file implements the per-request runtime: the message-processing loop that
// runs on a Request's own goroutines while its WebSocket is connected.
//
// [Request.process] is the select loop. Inbound client events are resolved by
// resolveEventFnCall and executed on the event goroutine via eventCaller;
// broadcasts and removal reports are applied by handleBroadcast and handleRemove;
// dirty elements are rendered into the outbound queue by getSendMsgs, sendQueue
// and makeUpdateList. onConnect runs the user ConnectFn once the socket is up.

import (
	"cmp"
	"errors"
	"fmt"
	"slices"
	"strings"

	"github.com/linkdata/jaws/lib/jid"
	"github.com/linkdata/jaws/lib/key"
	"github.com/linkdata/jaws/lib/what"
	"github.com/linkdata/jaws/lib/wire"
)

// process runs the main message-processing loop.
//
// It unsubscribes broadcastMsgCh and closes outboundMsgCh before returning. A
// loop panic is contained, a diagnostic is passed to [Jaws.Log], and the
// recovered value is returned unchanged for the caller to handle.
func (rq *Request) process(broadcastMsgCh chan wire.Message, incomingMsgCh <-chan wire.WsMsg, outboundMsgCh chan<- wire.WsMsg) (panicValue any) {
	jawsDoneCh := rq.Jaws.Done()
	// Snapshot cancelFn under rq.mu. getRequestLocked initializes it, and claim
	// layers it before process starts.
	rq.mu.RLock()
	httpDoneCh := rq.httpDoneCh
	cancelFn := rq.cancelFn
	rq.mu.RUnlock()
	eventDoneCh := make(chan struct{})
	eventCallCh := make(chan eventFnCall, cap(outboundMsgCh))
	go rq.eventCaller(eventCallCh, outboundMsgCh, eventDoneCh)

	defer func() {
		panicValue = recover()
		rq.Jaws.unsubscribe(broadcastMsgCh)
		// process runs only for a running Request, so its WebSocket earns grace.
		rq.killSession(true)
		cancelFn(nil)
		close(eventCallCh)
		for {
			select {
			case _, ok := <-incomingMsgCh:
				if !ok {
					incomingMsgCh = nil
				}
			case <-eventDoneCh:
				close(outboundMsgCh)
				if panicValue != nil {
					var err error
					var ok bool
					if err, ok = panicValue.(error); !ok {
						err = fmt.Errorf("jaws: %v panic: %v", rq, panicValue)
					}
					// Log non-fatally rather than MustLog: this runs in the cleanup
					// defer with no surrounding recover, and the request is torn down
					// regardless.
					_ = rq.Jaws.Log(err)
				}
				return
			}
		}
	}()

	for {
		var tagmsg wire.Message
		var wsmsg wire.WsMsg
		var ok bool

		rq.sendQueue(outboundMsgCh)

		// Drain pending dirty tags and exact Element targets, then call
		// JawsUpdate for the selected Elements. Updates queue browser messages
		// on the Request.
		for _, elem := range rq.makeUpdateList() {
			elem.JawsUpdate()
		}

		rq.sendQueue(outboundMsgCh)

		select {
		case <-jawsDoneCh:
		case <-httpDoneCh:
		case <-rq.Context().Done():
		case <-rq.wakeCh:
			continue
		case tagmsg, ok = <-broadcastMsgCh:
		case wsmsg, ok = <-incomingMsgCh:
			if ok {
				// incoming event message from the WebSocket
				rq.handleIncoming(wsmsg, eventCallCh)
				continue
			}
		}

		if !ok {
			// one of the channels are closed, so we're done
			return
		}

		rq.handleBroadcast(tagmsg, eventCallCh)
	}
}

// handleIncoming processes a single incoming WebSocket event message, queuing an
// event-function call or handling a child removal. Called only from process.
func (rq *Request) handleIncoming(wsmsg wire.WsMsg, eventCallCh chan eventFnCall) {
	if wsmsg.Jid.IsValid() {
		switch wsmsg.What {
		case what.Input, what.Click, what.ContextMenu, what.JsVar:
			rq.queueEvent(eventCallCh, rq.resolveEventFnCall(wsmsg.Jid, wsmsg.What, wsmsg.Data))
		case what.Remove:
			rq.handleRemove(wsmsg.Jid, wsmsg.Data)
		}
	}
}

// handleBroadcast processes a single broadcast (tag) message: it resolves the
// message destination to the affected elements and dispatches by command. Called
// only from process.
func (rq *Request) handleBroadcast(tagmsg wire.Message, eventCallCh chan eventFnCall) {
	// Reload, Redirect, Order and Alert are page-global commands: they apply to
	// the whole document, so emit the single Jid:0 frame and return before
	// resolving Dest.
	switch tagmsg.What {
	case what.Reload, what.Redirect, what.Order, what.Alert:
		rq.queue(wire.WsMsg{
			Jid:  0,
			Data: tagmsg.Data,
			What: tagmsg.What,
		})
		return
	case what.Call:
		// A nil or request-key destination selects the Request itself rather than
		// an Element. Serve has already filtered key-targeted messages, so the
		// empty Jid field tells the browser to perform one request-scoped call.
		switch tagmsg.Dest.(type) {
		case nil, key.Key:
			rq.queue(wire.WsMsg{
				Jid:  0,
				Data: tagmsg.Data,
				What: tagmsg.What,
			})
			return
		}
	}

	// collect all elements marked with the tag in the message
	var todo []*Element
	switch v := tagmsg.Dest.(type) {
	case nil:
		// matches no elements
	case key.Key:
		// request-targeted; page-global What values and Call already returned
		// above, so there are no elements to resolve here
	default:
		todo = rq.GetElements(v)
	}
	if tagmsg.What == what.Delete && len(todo) > 0 {
		rq.DeleteElements(todo)
	}

	for _, elem := range todo {
		switch tagmsg.What {
		case what.Delete:
			rq.queue(wire.WsMsg{
				Jid:  elem.Jid(),
				What: what.Delete,
			})
		case what.Input, what.Click, what.ContextMenu:
			// Input, Click or ContextMenu messages received here come from broadcasts;
			// primarily used in tests by injecting a wire.WsMsg on the inbound channel.
			// they won't be sent out on the WebSocket, but will queue up a
			// call to the event function (if any).
			// primary usecase is tests.
			rq.queueEvent(eventCallCh, rq.resolveEventFnCall(elem.Jid(), tagmsg.What, tagmsg.Data))
		case what.Hook:
			// Hook messages synchronously invoke the element's event handler; see
			// [what.Hook]. They exist for testing: the JaWS client never sends Hook,
			// so they only arrive here via Broadcast. The handler must not send any
			// messages itself. Returned errors follow the event alert policy.
			if err := rq.Jaws.Log(rq.callAllEventHandlers(elem.Jid(), tagmsg.What, tagmsg.Data)); err != nil && !errors.Is(err, ErrEventLogOnly) {
				var m wire.WsMsg
				m.FillAlert(err)
				m.Jid = elem.Jid()
				rq.queue(m)
			}
		case what.Update:
			elem.JawsUpdate()
		default:
			rq.queue(wire.WsMsg{
				Data: tagmsg.Data,
				Jid:  elem.Jid(),
				What: tagmsg.What,
			})
		}
	}
}

func (rq *Request) handleRemove(containerJid Jid, data string) {
	// Incoming what.Remove messages from jaws.js are cleanup acknowledgements sent
	// while applying server-driven DOM mutation commands (Inner, Replace, Delete or
	// Remove). Data is a tab-separated list of managed descendant IDs that were
	// removed from the DOM. The WebSocket Jid identifies the parent/container being
	// mutated and must not itself be deleted here.
	//
	// The client is already trusted only within its own request: a malicious client
	// can fully control the DOM and UI it presents to its user. Treating arbitrary
	// child removals as request-local state cleanup is therefore not a server-side
	// privilege boundary; IDs are only looked up in this Request.
	if containerJid > 0 {
		rq.mu.Lock()
		defer rq.mu.Unlock()
		// Mark removals immediately so all readers stop seeing them. Compact when
		// tombstones reach half the registry; fixed maintenance purges any remainder.
		// The threshold bounds retention and amortizes scans when a peer splits
		// acknowledgements across messages. Dirty/update activity never compacts, so
		// it cannot trigger a full scan per removal.
		for jidstr := range strings.SplitSeq(data, "\t") {
			if id := jid.ParseString(jidstr); id != containerJid {
				rq.markElementDeletedLocked(rq.getElementByJidLocked(id))
			}
		}
		if 2*rq.deletedElems >= len(rq.elems) {
			rq.purgeDeletedElementsLocked()
		}
	}
}

// wake schedules a processing pass without blocking. One pending signal is enough
// because process drains all queued work before waiting again.
func (rq *Request) wake() {
	select {
	case rq.wakeCh <- struct{}{}:
	default:
	}
}

// queue appends a single outbound message and wakes the processing loop under
// muQueue, the leaf lock that orders writes independently of rq.mu. The Request's
// process loop drains it through sendQueue and getSendMsgs; drainTailScript drains
// tail fixups.
func (rq *Request) queue(msg wire.WsMsg) {
	rq.muQueue.Lock()
	rq.wsQueue = append(rq.wsQueue, msg)
	if len(rq.wsQueue) == 1 {
		rq.wake()
	}
	rq.muQueue.Unlock()
}

// resolveEventFnCall resolves a single incoming event to its eligible target
// Elements. A zero id with a Click or ContextMenu carries a tab-separated list of
// bubbled element jids, whose live targets are resolved once each in
// first-occurrence order; any other id resolves to a single element. Only live,
// frozen Elements are selected, so the atomic frozen load publishes the completed
// handler slice before its later lock-free read.
//
// Resolution happens before the call enters the event FIFO, fixing target
// eligibility at acceptance. Later removal from any source does not cancel an
// accepted event, while an Element removed first cannot be selected. rq.mu
// protects target lookup; handlers run later without it.
func (rq *Request) resolveEventFnCall(id Jid, wht what.What, value string) (call eventFnCall) {
	call.wht = wht
	call.data = value
	rq.mu.RLock()
	if id == 0 {
		if wht == what.Click || wht == what.ContextMenu {
			const linearDedupLimit = 32
			var after string
			var found bool
			var seenMap map[*Element]struct{}
			call.data, after, found = strings.Cut(value, "\t")
			for found {
				var jidStr string
				jidStr, after, found = strings.Cut(after, "\t")
				if id = jid.ParseString(jidStr); id > 0 {
					if e := rq.getElementByJidLocked(id); e != nil && e.frozen.Load() {
						duplicate := e == call.elem
						if !duplicate {
							if seenMap == nil {
								duplicate = slices.Contains(call.more, e)
							} else {
								_, duplicate = seenMap[e]
							}
						}
						if duplicate {
							continue
						}
						if seenMap == nil && call.elem != nil && 1+len(call.more) == linearDedupLimit && after != "" {
							// Avoid separate deduplication storage for ordinary routes,
							// then switch before a client-controlled deep route becomes quadratic.
							seenMap = make(map[*Element]struct{}, linearDedupLimit+1)
							seenMap[call.elem] = struct{}{}
							for _, seenElem := range call.more {
								seenMap[seenElem] = struct{}{}
							}
						}
						if seenMap != nil {
							seenMap[e] = struct{}{}
						}
						switch {
						case call.elem == nil:
							call.elem = e
						case call.more == nil:
							remainingCap := 1
							if after != "" {
								remainingCap += 1 + strings.Count(after, "\t")
							}
							// Bound speculative allocation from the client-controlled route.
							call.more = make([]*Element, 0, min(remainingCap, 8))
							call.more = append(call.more, e)
						default:
							call.more = append(call.more, e)
						}
					}
				}
			}
		}
	} else {
		if e := rq.getElementByJidLocked(id); e != nil && e.frozen.Load() {
			call.elem = e
		}
	}
	rq.mu.RUnlock()
	return
}

// invoke calls the resolved Elements in order and returns the first result that
// is not ErrEventUnhandled. ErrEventUnhandled is normalized to nil.
func (call eventFnCall) invoke() (err error) {
	if call.elem != nil {
		err = callEventHandlers(call.elem.UI(), call.elem, call.more, call.wht, call.data)
	}
	if errors.Is(err, ErrEventUnhandled) {
		err = nil
	}
	return
}

// callAllEventHandlers resolves and dispatches a single event synchronously.
func (rq *Request) callAllEventHandlers(id Jid, wht what.What, value string) (err error) {
	err = rq.resolveEventFnCall(id, wht, value).invoke()
	return
}

// queueEvent hands a resolved event-function call to the eventCaller goroutine.
//
// eventCallCh is buffered to the outbound capacity; if it is full the request has
// fallen too far behind to stay consistent (an event would be lost), so it is
// cancelled rather than dropping the event, mirroring the broadcast back-pressure
// path in [Jaws.ServeWithTimeout]. cancel takes rq.mu, which the process loop does
// not hold when calling this. Calls without a resolved target are ignored.
func (rq *Request) queueEvent(eventCallCh chan eventFnCall, call eventFnCall) {
	if call.elem == nil {
		return
	}
	select {
	case eventCallCh <- call:
	default:
		targets := 1 + len(call.more)
		rq.cancel(fmt.Errorf("%w: %v: eventCallCh full sending FirstTarget=%v Targets=%d What=%v Data=%q", ErrRequestOverloaded, rq, call.elem.Jid(), targets, call.wht, call.data))
	}
}

// getSendMsgs drains the pending wsQueue, dropping messages addressed to elements
// that are not present (non-element messages and Delete are always kept), and
// returns the survivors sorted by Jid. It takes rq.mu (read) then muQueue, the
// order required by the lock hierarchy documented in jaws.go.
//
// what.Order is page-global (the browser ignores its Jid; see jawsPerform), but
// Element.Order queues it carrying the issuing element's Jid so the stable Jid sort
// places it after that element's own Append frames. It is therefore kept regardless
// of whether the issuer is still present, like the other page-global commands.
func (rq *Request) getSendMsgs() (toSend []wire.WsMsg) {
	rq.mu.RLock()
	defer rq.mu.RUnlock()

	rq.muQueue.Lock()
	defer rq.muQueue.Unlock()
	if len(rq.wsQueue) > 0 {
		for i := range rq.wsQueue {
			ok := rq.wsQueue[i].Jid < 1 || rq.wsQueue[i].What == what.Delete || rq.wsQueue[i].What == what.Order
			if !ok {
				ok = rq.getElementByJidLocked(rq.wsQueue[i].Jid) != nil
			}
			if ok {
				toSend = append(toSend, rq.wsQueue[i])
			}
		}
		// Zero the drained entries before reslicing so their Data payloads (full
		// Inner/Replace/Append HTML) are released promptly rather than lingering in the
		// backing array's unused capacity for the rest of the request (and into the
		// pooled buffer after teardown).
		clear(rq.wsQueue)
		rq.wsQueue = rq.wsQueue[:0]
	}

	slices.SortStableFunc(toSend, func(a, b wire.WsMsg) int { return cmp.Compare(a.Jid, b.Jid) })
	return
}

// sendQueue writes the drained outbound queue to outboundMsgCh, abandoning a send
// if the request context is cancelled.
func (rq *Request) sendQueue(outboundMsgCh chan<- wire.WsMsg) {
	msgs := rq.getSendMsgs()
	if len(msgs) == 0 {
		return
	}
	// Snapshot the done channel once for the whole batch. getSendMsgs already
	// froze the batch under a single lock, and rq.Context() takes rq.mu.RLock on
	// every call, so reading it per message would re-lock rq.mu K times during a
	// drain burst. A SetContext mid-drain is already unsynchronized relative to an
	// in-flight send, so capturing once changes no guaranteed behavior.
	done := rq.Context().Done()
	for _, msg := range msgs {
		select {
		case <-done:
		case outboundMsgCh <- msg:
		}
	}
}

// markElementDeletedLocked marks elem deleted and records its tombstone. Caller must
// hold rq.mu.
func (rq *Request) markElementDeletedLocked(elem *Element) {
	if elem != nil && elem.Request == rq && !elem.deleted.Load() {
		elem.deleted.Store(true)
		rq.deletedElems++
	}
}

// purgeDeletedElementsLocked drops deleted elements from the request's pending exact
// targets, element list, and tag entries, deleting tag entries that become empty.
//
// slices.DeleteFunc zeros the freed tail slots, so the dropped *Element pointers do
// not linger in the backing arrays. Caller must hold rq.mu.
func (rq *Request) purgeDeletedElementsLocked() {
	if rq.deletedElems == 0 {
		return
	}
	pred := func(e *Element) bool { return e.deleted.Load() }
	rq.todoDirt = slices.DeleteFunc(rq.todoDirt, func(tagValue any) bool {
		elem, ok := tagValue.(*Element)
		return ok && pred(elem)
	})
	rq.elems = slices.DeleteFunc(rq.elems, pred)
	for k := range rq.tagMap {
		rq.tagMap[k] = slices.DeleteFunc(rq.tagMap[k], pred)
		if len(rq.tagMap[k]) == 0 {
			delete(rq.tagMap, k)
		}
	}
	rq.deletedElems = 0
}

// DeleteElement removes elem from the [Request] element registry without
// queueing a browser operation.
//
// Use [Element.Remove] to remove a managed DOM child and unregister it together.
// DeleteElement is intended for elements that were never successfully rendered,
// or whose DOM lifecycle is managed separately.
//
// A nil elem is a no-op, matching [Request.Tag], [Request.TagExpanded] and
// [Request.TagsOf]; passing the nil that [Request.GetElementByJid] returns for an
// unknown Jid is therefore safe.
func (rq *Request) DeleteElement(elem *Element) {
	rq.mu.Lock()
	defer rq.mu.Unlock()
	rq.markElementDeletedLocked(elem)
	rq.purgeDeletedElementsLocked()
}

// DeleteElements removes all of elems from the [Request] element registry in a
// single pass, without queueing browser operations.
//
// It is the batched form of [Request.DeleteElement], for unregistering a whole
// subtree at once: one pass over the registry regardless of how many elements are
// dropped, rather than one pass per element. Nil elements and elements belonging to
// another Request are skipped, and repeated elements are tolerated.
func (rq *Request) DeleteElements(elems []*Element) {
	rq.mu.Lock()
	defer rq.mu.Unlock()
	for _, elem := range elems {
		rq.markElementDeletedLocked(elem)
	}
	rq.purgeDeletedElementsLocked()
}

// makeUpdateList drains exact Element targets and pending-dirt tags, resolves them
// to distinct Elements, and returns them sorted by Jid. It takes rq.mu. The Request
// processing loop calls JawsUpdate on each returned Element.
func (rq *Request) makeUpdateList() (todo []*Element) {
	rq.mu.Lock()
	seen := map[*Element]struct{}{}
	for _, tagValue := range rq.todoDirt {
		if elem, exact := tagValue.(*Element); exact {
			// appendDirtyTags establishes ownership and liveness. Browser-reported
			// deletion may leave a tombstone until amortized compaction.
			if _, ok := seen[elem]; !ok && !elem.deleted.Load() {
				seen[elem] = struct{}{}
				todo = append(todo, elem)
			}
			continue
		}
		for _, elem := range rq.tagMap[tagValue] {
			if _, ok := seen[elem]; !ok && !elem.deleted.Load() {
				seen[elem] = struct{}{}
				todo = append(todo, elem)
			}
		}
	}
	clear(rq.todoDirt)
	rq.todoDirt = rq.todoDirt[:0]
	rq.mu.Unlock()
	slices.SortFunc(todo, func(a, b *Element) int { return cmp.Compare(a.Jid(), b.Jid()) })
	return
}

// eventCaller calls event functions.
//
// Once the Request context is cancelled it stops invoking handlers and drains the
// remaining queued calls as no-ops. This cannot strand events on a still-live
// Request: process selects on the same rq.Context().Done() and returns, then closes
// eventCallCh, so the no-op drain is always part of teardown.
func (rq *Request) eventCaller(eventCallCh <-chan eventFnCall, outboundMsgCh chan<- wire.WsMsg, eventDoneCh chan<- struct{}) {
	defer close(eventDoneCh)
	for call := range eventCallCh {
		select {
		case <-rq.Context().Done():
			continue
		default:
		}
		if err := rq.Jaws.Log(call.invoke()); err != nil && !errors.Is(err, ErrEventLogOnly) {
			var m wire.WsMsg
			m.FillAlert(err)
			// This error alert is best-effort: unlike queueEvent, which cancels the
			// Request with ErrRequestOverloaded when its channel fills (dropping a
			// queued event could desync browser and backend state), a dropped alert
			// loses no state — the underlying error is already queued above — so a
			// full outbound channel here is queued for logging and the alert is
			// discarded rather than tearing down the Request.
			select {
			case outboundMsgCh <- m:
			default:
				_ = rq.Jaws.Log(fmt.Errorf("jaws: outboundMsgCh full sending event error '%s'", err.Error()))
			}
		}
	}
}

// onConnect calls the [Request]'s [ConnectFn] if it is not nil, returning its
// error or converting its panic to an error. Returns nil if [ConnectFn] is nil.
func (rq *Request) onConnect() (err error) {
	rq.mu.RLock()
	connectFn := rq.connectFn
	rq.mu.RUnlock()
	if connectFn != nil {
		defer func() {
			if x := recover(); x != nil {
				err = fmt.Errorf("jaws: %v ConnectFn panic: %v", rq, x)
			}
		}()
		err = connectFn(rq)
	}
	return
}