Reactor Core: signals, event bus, job scheduler, and coordinator kernel
Dieses Skript sollte nicht direkt installiert werden. Es handelt sich hier um eine Bibliothek für andere Skripte, welche über folgenden Befehl in den Metadaten eines Skriptes eingebunden wird // @require https://update.greasyfork.org/scripts/591444/1903232/Brazen%20Framework%20-%20Reactor.js
// ==UserScript==
// @name Brazen Framework - Reactor
// @namespace brazenvoid
// @version 0.0.1
// @author brazenvoid
// @license GPL-3.0-only
// @description Reactor Core: signals, event bus, job scheduler, and coordinator kernel
// @run-at document-end
// ==/UserScript==
// -------------------------------------------------------------------------
// Signals
// -------------------------------------------------------------------------
/**
* @typedef {() => void} Unsubscribe
*/
/**
* @typedef {Object} Atom
* @property {unknown} value
* @property {number} version
* @property {(next: unknown) => void} set
* @property {(fn: (value: unknown) => void) => Unsubscribe} subscribe
*/
/**
* @typedef {Object} Computed
* @property {unknown} value
* @property {number} version
* @property {(fn: (value: unknown) => void) => Unsubscribe} subscribe
*/
/**
* @typedef {Object} Effect
* @property {() => void} dispose
*/
class BrazenSignalCycleError extends Error
{
constructor(message = 'Signal dependency cycle detected')
{
super(message)
this.name = 'BrazenSignalCycleError'
}
}
/** @type {number} */
let propagationGeneration = 0
/** @type {number} */
let batchDepth = 0
/** @type {Set<InternalNode>} */
let pendingDirtyAtoms = new Set()
/**
* @typedef {Object} TrackingFrame
* @property {InternalConsumer|null} owner
* @property {Set<InternalNode>} deps
* @property {InternalComputed[]} stack
* @property {boolean} [forceFresh]
*/
/** @type {Set<InternalComputed>} */
const initializingComputeds = new Set()
/** @type {TrackingFrame|null} */
let trackingFrame = null
/**
* @typedef {Object} InternalNode
* @property {'atom'|'computed'} kind
* @property {Set<InternalConsumer>} dependents
*/
/**
* @typedef {Object} InternalConsumer
* @property {'computed'|'effect'} kind
* @property {Map<InternalNode, number>} depVersions
* @property {Set<InternalNode>} depNodes
* @property {Set<(value: unknown) => void>} subscribers
*/
/**
* @param {InternalConsumer} consumer
* @param {InternalNode} producer
*/
function linkDependency(consumer, producer)
{
if (!consumer.depVersions.has(producer)) {
consumer.depNodes.add(producer)
consumer.depVersions.set(producer, producer.version)
producer.dependents.add(consumer)
}
else if (trackingFrame) {
consumer.depVersions.set(producer, producer.version)
}
}
/**
* @param {InternalConsumer} consumer
*/
function clearConsumerDependencies(consumer)
{
for (const producer of consumer.depNodes) {
producer.dependents.delete(consumer)
}
consumer.depNodes.clear()
consumer.depVersions.clear()
}
/**
* @param {InternalComputed} computed
* @returns {boolean}
*/
function isComputedStale(computed)
{
for (const [dep, seenVersion] of computed.depVersions) {
if (dep.version !== seenVersion) {
return true
}
}
return false
}
/**
* @param {InternalComputed} computed
*/
function assertNoCycle(computed)
{
if (trackingFrame && trackingFrame.stack.includes(computed)) {
throw new BrazenSignalCycleError('Signal dependency cycle detected')
}
}
/**
* @typedef {InternalConsumer & {
* kind: 'computed',
* read: () => unknown,
* cachedValue: unknown,
* version: number,
* path?: string,
* }} InternalAtom
*/
/**
* @typedef {InternalNode & {
* kind: 'atom',
* path?: string,
* cachedValue: unknown,
* version: number,
* set: (next: unknown) => void,
* get value: () => unknown,
* subscribe: (fn: (value: unknown) => void) => Unsubscribe,
* }} InternalAtom
*/
/**
* @typedef {InternalConsumer & {
* kind: 'computed',
* read: () => unknown,
* cachedValue: unknown,
* version: number,
* name?: string,
* get value: () => unknown,
* subscribe: (fn: (value: unknown) => void) => Unsubscribe,
* }} InternalComputed
*/
/**
* @typedef {InternalConsumer & {
* kind: 'effect',
* run: () => void,
* disposed: boolean,
* cleanup: (() => void)|null,
* name?: string,
* }} InternalEffect
*/
/**
* @param {InternalComputed} computed
* @returns {unknown}
*/
function recomputeComputed(computed)
{
assertNoCycle(computed)
const previousFrame = trackingFrame
const frame = {
owner: computed,
deps: new Set(),
stack: previousFrame ? [...previousFrame.stack, computed] : [computed],
forceFresh: previousFrame?.forceFresh,
}
trackingFrame = frame
clearConsumerDependencies(computed)
let nextValue
try {
nextValue = computed.read()
}
finally {
trackingFrame = previousFrame
}
for (const dep of frame.deps) {
linkDependency(computed, dep)
}
computed.cachedValue = nextValue
computed.version = computed.depNodes.size === 0
? propagationGeneration
: [...computed.depNodes].reduce((max, dep) => Math.max(max, dep.version), 0)
return nextValue
}
/**
* @param {InternalComputed} computed
* @returns {unknown}
*/
function readComputedValue(computed)
{
assertNoCycle(computed)
if (trackingFrame?.owner && trackingFrame.owner !== computed) {
trackingFrame.deps.add(computed)
linkDependency(trackingFrame.owner, computed)
}
if (isComputedStale(computed) || trackingFrame?.forceFresh) {
recomputeComputed(computed)
}
return computed.cachedValue
}
/**
* @param {InternalAtom} atom
* @returns {unknown}
*/
function readAtomValue(atom)
{
if (trackingFrame && trackingFrame.owner) {
trackingFrame.deps.add(atom)
linkDependency(trackingFrame.owner, atom)
}
return atom.cachedValue
}
/**
* @param {Set<InternalAtom>} dirtyAtoms
*/
function propagateFromDirtyAtoms(dirtyAtoms)
{
if (dirtyAtoms.size === 0) {
return
}
propagationGeneration++
/** @type {Set<InternalComputed>} */
const affectedComputeds = new Set()
/** @type {InternalNode[]} */
const queue = [...dirtyAtoms]
while (queue.length > 0) {
const node = queue.pop()
for (const dependent of node.dependents) {
if (dependent.kind === 'computed') {
if (!affectedComputeds.has(dependent)) {
affectedComputeds.add(dependent)
queue.push(dependent)
}
}
}
}
const sortedComputeds = topoSortComputeds(affectedComputeds)
/** @type {Map<InternalConsumer, unknown>} */
const previousValues = new Map()
for (const computed of sortedComputeds) {
previousValues.set(computed, computed.cachedValue)
recomputeComputed(computed)
}
const affectedEffects = collectTransitiveEffects([...dirtyAtoms, ...sortedComputeds])
for (const effect of affectedEffects) {
if (effectNeedsRun(effect)) {
runEffect(effect)
}
}
for (const atom of dirtyAtoms) {
notifySubscribers(atom, atom.cachedValue)
}
for (const computed of sortedComputeds) {
if (!Object.is(previousValues.get(computed), computed.cachedValue)) {
notifySubscribers(computed, computed.cachedValue)
}
}
}
/**
* @param {InternalNode[]} roots
* @returns {Set<InternalEffect>}
*/
function collectTransitiveEffects(roots)
{
/** @type {Set<InternalEffect>} */
const effects = new Set()
/** @type {InternalNode[]} */
const queue = [...roots]
/** @type {Set<InternalNode>} */
const visited = new Set()
while (queue.length > 0) {
const node = queue.pop()
if (!node || visited.has(node)) {
continue
}
visited.add(node)
for (const dependent of node.dependents) {
if (dependent.kind === 'effect' && !dependent.disposed) {
effects.add(dependent)
}
else if (dependent.kind === 'computed') {
queue.push(dependent)
}
}
}
return effects
}
/**
* @param {Set<InternalComputed>} computeds
* @returns {InternalComputed[]}
*/
function topoSortComputeds(computeds)
{
/** @type {InternalComputed[]} */
const sorted = []
/** @type {Set<InternalComputed>} */
const visited = new Set()
/** @type {Set<InternalComputed>} */
const visiting = new Set()
/**
* @param {InternalComputed} computed
*/
function visit(computed)
{
if (visited.has(computed) || !computeds.has(computed)) {
return
}
if (visiting.has(computed)) {
throw new BrazenSignalCycleError('Signal dependency cycle detected')
}
visiting.add(computed)
for (const dep of computed.depNodes) {
if (dep.kind === 'computed') {
visit(dep)
}
}
visiting.delete(computed)
visited.add(computed)
sorted.push(computed)
}
for (const computed of computeds) {
visit(computed)
}
return sorted
}
/**
* @param {InternalEffect} effect
* @returns {boolean}
*/
function effectNeedsRun(effect)
{
for (const [dep, seenVersion] of effect.depVersions) {
if (dep.version !== seenVersion) {
return true
}
}
return false
}
/**
* @param {InternalEffect} effect
*/
function runEffect(effect)
{
if (effect.disposed) {
return
}
if (effect.cleanup) {
effect.cleanup()
effect.cleanup = null
}
const previousFrame = trackingFrame
const frame = {
owner: effect,
deps: new Set(),
stack: previousFrame ? [...previousFrame.stack] : [],
forceFresh: previousFrame?.forceFresh,
}
trackingFrame = frame
clearConsumerDependencies(effect)
try {
const result = effect.run()
if (typeof result === 'function') {
effect.cleanup = result
}
}
finally {
trackingFrame = previousFrame
}
for (const dep of frame.deps) {
linkDependency(effect, dep)
}
}
/**
* @param {InternalConsumer} node
* @param {unknown} value
*/
function notifySubscribers(node, value)
{
for (const fn of node.subscribers) {
fn(value)
}
}
/**
* @param {InternalAtom} atom
*/
function markAtomDirty(atom)
{
if (batchDepth > 0) {
pendingDirtyAtoms.add(atom)
return
}
propagateFromDirtyAtoms(new Set([atom]))
}
/** @type {Map<string, InternalAtom>} */
const pathRegistry = new Map()
/**
* @param {InternalAtom} atom
* @param {unknown | ((prev: unknown) => unknown)} next
* @return {number}
*/
function commitAtomWrite(atom, next)
{
let resolved = typeof next === 'function' ? next(atom.cachedValue) : next
if (Object.is(atom.cachedValue, resolved)) {
return atom.version
}
atom.cachedValue = resolved
atom.version++
markAtomDirty(atom)
return atom.version
}
/**
* @param {string} path
* @param {unknown} initial
* @return {InternalAtom}
*/
function getOrCreatePathAtom(path, initial)
{
let existing = pathRegistry.get(path)
if (existing) {
return existing
}
/** @type {InternalAtom} */
let atom = {
kind: 'atom',
path,
cachedValue: initial,
version: 0,
dependents: new Set(),
subscribers: new Set(),
set(next) {
commitAtomWrite(atom, next)
},
get value() {
return readAtomValue(atom)
},
subscribe(fn) {
atom.subscribers.add(fn)
return () => {
atom.subscribers.delete(fn)
}
},
}
pathRegistry.set(path, atom)
return atom
}
/**
* @template T
* @param {T} initial
* @param {{ path?: string }} [options]
* @returns {Atom<T>}
*/
function createAtom(initial, options = {})
{
if (options.path) {
return getOrCreatePathAtom(options.path, initial)
}
/** @type {InternalAtom} */
const atom = {
kind: 'atom',
path: options.path,
cachedValue: initial,
version: 0,
dependents: new Set(),
subscribers: new Set(),
set(next) {
commitAtomWrite(atom, next)
},
get value() {
return readAtomValue(atom)
},
subscribe(fn) {
atom.subscribers.add(fn)
return () => {
atom.subscribers.delete(fn)
}
},
}
return atom
}
/**
* @template T
* @param {() => T} read
* @param {{ name?: string }} [options]
* @returns {Computed<T>}
*/
function createComputed(read, options = {})
{
/** @type {InternalComputed} */
const computed = {
kind: 'computed',
name: options.name,
read,
cachedValue: undefined,
version: 0,
dependents: new Set(),
depNodes: new Set(),
depVersions: new Map(),
subscribers: new Set(),
get value() {
return readComputedValue(computed)
},
subscribe(fn) {
computed.subscribers.add(fn)
return () => {
computed.subscribers.delete(fn)
}
},
}
initializingComputeds.add(computed)
try {
recomputeComputed(computed)
}
finally {
initializingComputeds.delete(computed)
}
return computed
}
/**
* @param {() => void | (() => void)} fn
* @param {{ name?: string }} [options]
* @returns {Effect}
*/
function createEffect(fn, options = {})
{
/** @type {InternalEffect} */
const effect = {
kind: 'effect',
name: options.name,
run: fn,
disposed: false,
cleanup: null,
dependents: new Set(),
depNodes: new Set(),
depVersions: new Map(),
subscribers: new Set(),
}
runEffect(effect)
return {
dispose() {
if (effect.disposed) {
return
}
effect.disposed = true
if (effect.cleanup) {
effect.cleanup()
effect.cleanup = null
}
clearConsumerDependencies(effect)
},
}
}
/**
* @param {() => void} fn
*/
function batch(fn)
{
batchDepth++
try {
fn()
}
finally {
batchDepth--
if (batchDepth === 0 && pendingDirtyAtoms.size > 0) {
const dirty = pendingDirtyAtoms
pendingDirtyAtoms = new Set()
propagateFromDirtyAtoms(dirty)
}
}
}
/**
* @returns {number}
*/
function getPropagationGeneration()
{
return propagationGeneration
}
/**
* @param {() => unknown} [read]
*/
function assertAcyclic(read)
{
if (typeof read !== 'function') {
return
}
const previousFrame = trackingFrame
trackingFrame = {
owner: null,
deps: new Set(),
stack: [],
forceFresh: true,
}
try {
read()
}
finally {
trackingFrame = previousFrame
}
}
/**
* @template T
* @param {string} path
* @param {T} initial
*/
function atom(path, initial)
{
let internal = getOrCreatePathAtom(path, initial)
return {
path,
read: () => readAtomValue(internal),
write: (next) => commitAtomWrite(internal, next),
peekVersion: () => internal.version,
get value() {
return readAtomValue(internal)
},
get version() {
return internal.version
},
set(next) {
commitAtomWrite(internal, next)
},
subscribe(fn) {
return internal.subscribe(fn)
},
}
}
function replicaAtom(path, initial)
{
return atom(path, initial)
}
function computed(path, deps, fn)
{
let internal = createComputed(() => {
if (typeof deps === 'function') {
deps()
}
return fn()
}, {name: path})
return {
path,
read: () => readComputedValue(internal),
peekVersion: () => internal.version,
subscribe(fn) {
return internal.subscribe(fn)
},
}
}
function effect(deps, fn)
{
return createEffect(() => {
if (typeof deps === 'function') {
deps()
}
return fn()
}, {name: 'effect'})
}
function applyPatches(patches)
{
if (!Array.isArray(patches)) {
return
}
batch(() => {
for (const patch of patches) {
if (!patch?.path || typeof patch.version !== 'number') {
continue
}
let internal = pathRegistry.get(patch.path)
if (!internal) {
internal = getOrCreatePathAtom(patch.path, patch.value)
}
if (patch.version <= internal.version) {
continue
}
internal.cachedValue = patch.value
internal.version = patch.version
markAtomDirty(internal)
}
})
}
function markDependency(target)
{
if (!trackingFrame?.owner || !target) {
return
}
if (typeof target.read === 'function') {
target.read()
} else if ('value' in target) {
void target.value
}
}
const BrazenSignals = Object.freeze({
atom,
replicaAtom,
computed,
effect,
batch,
applyPatches,
markDependency,
assertAcyclic,
BrazenSignalCycleError,
createAtom,
createComputed,
createEffect,
getPropagationGeneration,
})
globalThis.BrazenSignals = BrazenSignals
globalThis.createAtom = createAtom
globalThis.createComputed = createComputed
globalThis.createEffect = createEffect
globalThis.batch = batch
globalThis.getPropagationGeneration = getPropagationGeneration
globalThis.assertAcyclic = assertAcyclic
globalThis.BrazenSignalCycleError = BrazenSignalCycleError
// -------------------------------------------------------------------------
// EventBus
// -------------------------------------------------------------------------
/** @typedef {string} TabId */
/** @typedef {() => void} Unsubscribe */
/**
* @typedef {object} SignalPatch
* @property {string} path
* @property {number} version
* @property {unknown} value
*/
/**
* @typedef {object} BusMessageBase
* @property {'command' | 'patch' | 'snapshot-request' | 'snapshot-response'} kind
* @property {TabId} tabId
* @property {number} [seq]
* @property {number} [ts]
*/
/**
* @typedef {object} CommandMessage
* @property {'command'} kind
* @property {TabId} tabId
* @property {object} command
*/
/**
* @typedef {object} PatchMessage
* @property {'patch'} kind
* @property {TabId} tabId
* @property {number} seq
* @property {SignalPatch[]} patches
* @property {object[]} [events]
*/
/**
* @typedef {object} SnapshotRequestMessage
* @property {'snapshot-request'} kind
* @property {TabId} tabId
* @property {string} requestId
* @property {number} sinceSeq
*/
/**
* @typedef {object} SnapshotResponseMessage
* @property {'snapshot-response'} kind
* @property {TabId} tabId
* @property {string} requestId
* @property {Record<string, unknown>} snapshot
* @property {number} snapshotSeq
* @property {SignalPatch[]} [catchUpPatches]
*/
/** @typedef {CommandMessage | PatchMessage | SnapshotRequestMessage | SnapshotResponseMessage} BusMessage */
/**
* @typedef {object} SnapshotResponderResult
* @property {Record<string, unknown>} snapshot
* @property {number} snapshotSeq
* @property {SignalPatch[]} [catchUpPatches]
*/
/**
* @callback SnapshotResponder
* @param {SnapshotRequestMessage} request
* @return {SnapshotResponderResult | Promise<SnapshotResponderResult>}
*/
/**
* @callback BroadcastChannelFactory
* @param {string} channelName
* @return {{ postMessage: (data: unknown) => void, close: () => void, onmessage: ((event: { data: unknown }) => void) | null }}
*/
/**
* @typedef {object} BrazenEventBusOptions
* @property {TabId} [tabId]
* @property {SnapshotResponder} [onSnapshotRequest]
* @property {number} [snapshotTimeoutMs]
* @property {BroadcastChannelFactory} [createBroadcastChannel]
*/
const DEFAULT_SNAPSHOT_TIMEOUT_MS = 5000
const MAX_DELIVERY_QUEUE = 4096
const GAP_RESYNC_WAIT_MS = 250
class BrazenEventBus
{
// -------------------------------------------------------------------------
// Static public methods
// -------------------------------------------------------------------------
/**
* @param {string} scriptPrefix
* @param {BrazenEventBusOptions} [options]
* @return {BrazenEventBus}
*/
static create(scriptPrefix, options = {})
{
return new BrazenEventBus(scriptPrefix, options)
}
// -------------------------------------------------------------------------
// Public instance fields
// -------------------------------------------------------------------------
/** @type {TabId} */
tabId
/** @type {string} */
channelName
// -------------------------------------------------------------------------
// Protected class variables
// -------------------------------------------------------------------------
/** @type {ReturnType<BroadcastChannelFactory> | null} */
_channel = null
/** @type {Set<(message: BusMessage) => void>} */
_localListeners = new Set()
/** @type {Set<(message: BusMessage) => void>} */
_subscribers = new Set()
/** @type {number} */
_seq = 0
/** @type {boolean} */
_disposed = false
/** @type {boolean} */
_snapshotReady = false
/** @type {ReturnType<typeof setTimeout>|null} */
_gapResyncTimer = null
/** @type {boolean} */
_gapResyncInFlight = false
/** @type {number} */
_nextExpectedPatchSeq = 0
/** @type {PatchMessage[]} */
_patchBuffer = []
/** @type {BusMessage[]} */
_deliveryQueue = []
/** @type {boolean} */
_delivering = false
/**
* @type {Map<string, { resolve: (message: SnapshotResponseMessage) => void, reject: (error: Error) => void, timer: ReturnType<typeof setTimeout> }>}
*/
_pendingSnapshots = new Map()
/** @type {SnapshotResponder | null} */
_snapshotResponder = null
/** @type {number} */
_snapshotTimeoutMs = DEFAULT_SNAPSHOT_TIMEOUT_MS
/** @type {BroadcastChannelFactory} */
_createBroadcastChannel
// -------------------------------------------------------------------------
// Constructor
// -------------------------------------------------------------------------
/**
* @param {string} scriptPrefix
* @param {BrazenEventBusOptions} [options]
*/
constructor(scriptPrefix, options = {})
{
this.tabId = options.tabId ?? crypto.randomUUID()
this.channelName = `brazen-${scriptPrefix}`
this._snapshotResponder = options.onSnapshotRequest ?? null
this._snapshotTimeoutMs = options.snapshotTimeoutMs ?? DEFAULT_SNAPSHOT_TIMEOUT_MS
this._createBroadcastChannel = options.createBroadcastChannel ?? ((channelName) => new BroadcastChannel(channelName))
if (this._snapshotResponder) {
this._snapshotReady = true
}
this._openChannel()
}
// -------------------------------------------------------------------------
// Public instance methods
// -------------------------------------------------------------------------
/**
* In-tab + cross-tab publish. Followers: command only. Coordinator: all kinds.
* @param {BusMessage} message
*/
publish(message)
{
this._assertActive()
const stamped = this._stamp(message)
this._emitLocal(stamped)
this._postToChannel(stamped)
}
/**
* In-tab only publish (does not cross BroadcastChannel).
* @param {BusMessage} message
*/
emitLocal(message)
{
this._assertActive()
this._emitLocal(this._stamp(message))
}
/**
* Subscribe to in-tab events from publish/emitLocal only.
* @param {(message: BusMessage) => void} handler
* @return {Unsubscribe}
*/
onLocal(handler)
{
this._assertActive()
this._localListeners.add(handler)
return () => {
this._localListeners.delete(handler)
}
}
/**
* Ordered delivery per tab: commands FIFO; patches in seq order.
* @param {(message: BusMessage) => void} handler
* @return {Unsubscribe}
*/
subscribe(handler)
{
this._assertActive()
this._subscribers.add(handler)
return () => {
this._subscribers.delete(handler)
}
}
/**
* Coordinator: assign next seq (strictly monotonic with overflow guard).
* @return {number}
*/
nextSeq()
{
this._assertActive()
if (this._seq >= Number.MAX_SAFE_INTEGER) {
this._seq = 0
this._nextExpectedPatchSeq = 0
this._deliveryQueue = []
this._patchBuffer = []
return 1
}
this._seq += 1
return this._seq
}
/**
* Follower cold start: request full snapshot before subscribing to patch stream.
* @param {number} [sinceSeq]
* @return {Promise<SnapshotResponseMessage>}
*/
requestSnapshot(sinceSeq = 0)
{
this._assertActive()
const requestId = crypto.randomUUID()
this._snapshotReady = false
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
this._pendingSnapshots.delete(requestId)
reject(new Error('BrazenEventBus: snapshot request timed out'))
}, this._snapshotTimeoutMs)
this._pendingSnapshots.set(requestId, {resolve, reject, timer})
this._postToChannel({
kind: 'snapshot-request',
tabId: this.tabId,
requestId,
sinceSeq,
})
})
}
/**
* Coordinator: respond to snapshot-request.
* @param {SnapshotRequestMessage} request
* @return {Promise<void>}
*/
async handleSnapshotRequest(request)
{
this._assertActive()
if (!this._snapshotResponder) {
throw new Error('BrazenEventBus: no snapshot responder registered')
}
const result = await this._snapshotResponder(request)
this.publish({
kind: 'snapshot-response',
tabId: this.tabId,
requestId: request.requestId,
snapshot: structuredClone(result.snapshot),
snapshotSeq: result.snapshotSeq,
catchUpPatches: result.catchUpPatches ? structuredClone(result.catchUpPatches) : undefined,
})
}
/**
* Tear down channel + listeners.
*/
dispose()
{
if (this._disposed) {
return
}
this._disposed = true
for (const pending of this._pendingSnapshots.values()) {
clearTimeout(pending.timer)
pending.reject(new Error('BrazenEventBus disposed'))
}
this._pendingSnapshots.clear()
this._localListeners.clear()
this._subscribers.clear()
this._deliveryQueue = []
this._patchBuffer = []
if (this._gapResyncTimer) {
clearTimeout(this._gapResyncTimer)
this._gapResyncTimer = null
}
if (this._channel) {
this._channel.onmessage = null
this._channel.close()
this._channel = null
}
}
// -------------------------------------------------------------------------
// Private class methods
// -------------------------------------------------------------------------
/**
* @private
*/
_assertActive()
{
if (this._disposed) {
throw new Error('BrazenEventBus disposed')
}
}
/**
* @private
*/
_openChannel()
{
this._channel = this._createBroadcastChannel(this.channelName)
this._channel.onmessage = (event) => {
const message = /** @type {BusMessage} */ (event.data)
if (!message || message.tabId === this.tabId) {
return
}
this._receiveRemote(message)
}
}
/**
* @param {BusMessage} message
* @private
*/
_stamp(message)
{
return structuredClone({
...message,
tabId: message.tabId ?? this.tabId,
ts: message.ts ?? Date.now(),
})
}
/**
* @param {BusMessage} message
* @private
*/
_emitLocal(message)
{
for (const listener of this._localListeners) {
listener(message)
}
this._enqueueForSubscribers(message, {bypassSnapshotGate: true})
}
/**
* @param {BusMessage} message
* @private
*/
_postToChannel(message)
{
if (this._channel) {
this._channel.postMessage(structuredClone(message))
}
}
/**
* @param {BusMessage} message
* @private
*/
_receiveRemote(message)
{
if (message.kind === 'snapshot-response') {
this._tryResolveSnapshot(/** @type {SnapshotResponseMessage} */ (message))
}
if (message.kind === 'snapshot-request' && this._snapshotResponder) {
void this.handleSnapshotRequest(/** @type {SnapshotRequestMessage} */ (message))
}
this._enqueueForSubscribers(message)
}
/**
* @param {BusMessage} message
* @param {{bypassSnapshotGate?: boolean}} [options]
* @private
*/
_enqueueForSubscribers(message, options = {})
{
if (!this._snapshotReady && message.kind === 'patch' && !options.bypassSnapshotGate) {
this._patchBuffer.push(/** @type {PatchMessage} */ (message))
return
}
if (this._deliveryQueue.length >= MAX_DELIVERY_QUEUE) {
this._deliveryQueue.shift()
}
this._deliveryQueue.push(message)
this._drainDeliveryQueue()
}
/**
* @private
*/
_drainDeliveryQueue()
{
if (this._delivering) {
return
}
this._delivering = true
try {
while (this._deliveryQueue.length > 0) {
const index = this._findNextDeliverableIndex()
if (index < 0) {
this._maybeScheduleGapResync()
break
}
const message = this._deliveryQueue.splice(index, 1)[0]
this._deliverToSubscribers(message)
}
} finally {
this._delivering = false
}
}
/**
* @return {number}
* @private
*/
_findNextDeliverableIndex()
{
for (let index = 0; index < this._deliveryQueue.length; index += 1) {
const message = this._deliveryQueue[index]
if (message.kind !== 'patch') {
return index
}
const patch = /** @type {PatchMessage} */ (message)
if (patch.seq <= this._nextExpectedPatchSeq) {
return index
}
if (patch.seq === this._nextExpectedPatchSeq + 1) {
return index
}
}
return -1
}
/**
* @private
*/
_maybeScheduleGapResync()
{
if (this._snapshotResponder || this._gapResyncInFlight || this._gapResyncTimer) {
return
}
let minGapSeq = Infinity
for (const message of this._deliveryQueue) {
if (message.kind === 'patch') {
let seq = /** @type {PatchMessage} */ (message).seq
if (seq > this._nextExpectedPatchSeq + 1) {
minGapSeq = Math.min(minGapSeq, seq)
}
}
}
if (minGapSeq === Infinity) {
return
}
this._gapResyncTimer = setTimeout(() => {
this._gapResyncTimer = null
this._gapResyncInFlight = true
void this.requestSnapshot(this._nextExpectedPatchSeq)
.then(() => {
this._deliveryQueue = []
this._patchBuffer = []
})
.catch(() => {})
.finally(() => {
this._gapResyncInFlight = false
})
}, GAP_RESYNC_WAIT_MS)
}
/**
* @param {BusMessage} message
* @private
*/
_deliverToSubscribers(message)
{
if (message.kind === 'patch') {
const patch = /** @type {PatchMessage} */ (message)
if (patch.seq <= this._nextExpectedPatchSeq) {
return
}
this._nextExpectedPatchSeq = patch.seq
}
for (const subscriber of this._subscribers) {
subscriber(message)
}
}
/**
* @param {SnapshotResponseMessage} message
* @private
*/
_tryResolveSnapshot(message)
{
const pending = this._pendingSnapshots.get(message.requestId)
if (!pending) {
return
}
clearTimeout(pending.timer)
this._pendingSnapshots.delete(message.requestId)
this._snapshotReady = true
this._nextExpectedPatchSeq = message.snapshotSeq
pending.resolve(message)
this._releaseBufferedPatches()
}
/**
* @private
*/
_releaseBufferedPatches()
{
if (this._patchBuffer.length === 0) {
return
}
this._patchBuffer.sort((left, right) => left.seq - right.seq)
const buffered = this._patchBuffer
this._patchBuffer = []
for (const patch of buffered) {
if (patch.seq > this._nextExpectedPatchSeq) {
this._deliveryQueue.push(patch)
}
}
this._drainDeliveryQueue()
}
}
globalThis.BrazenEventBus = BrazenEventBus
// -------------------------------------------------------------------------
// JobRegistry
// -------------------------------------------------------------------------
/** @typedef {import('./reactor-core.spec.md').JobDescriptor} JobDescriptor */
/** @type {string} */
const JOB_TYPE_RESOLVE = 'resolve'
/** @type {string} */
const JOB_TYPE_DOWNLOAD = 'download'
class BrazenJobRegistry
{
// -------------------------------------------------------------------------
// Protected class variables
// -------------------------------------------------------------------------
/** @type {Map<string, JobDescriptor>} */
_descriptors = new Map()
// -------------------------------------------------------------------------
// Public class methods
// -------------------------------------------------------------------------
/**
* Register a job type descriptor.
* @template TPayload
* @template TResult
* @param {string} type
* @param {JobDescriptor<TPayload, TResult>} descriptor
*/
register(type, descriptor)
{
if (!type || typeof type !== 'string') {
throw new Error('BrazenJobRegistry.register: type must be a non-empty string')
}
if (!descriptor || typeof descriptor.run !== 'function') {
throw new Error(`BrazenJobRegistry.register: "${type}" descriptor.run must be a function`)
}
this._descriptors.set(type, descriptor)
}
/**
* @param {string} type
* @return {JobDescriptor | undefined}
*/
get(type)
{
return this._descriptors.get(type)
}
/**
* @return {string[]}
*/
types()
{
return [...this._descriptors.keys()]
}
}
globalThis.BrazenJobRegistry = BrazenJobRegistry
globalThis.JOB_TYPE_RESOLVE = JOB_TYPE_RESOLVE
globalThis.JOB_TYPE_DOWNLOAD = JOB_TYPE_DOWNLOAD
// -------------------------------------------------------------------------
// Scheduler
// -------------------------------------------------------------------------
/**
* @typedef {object} SchedulerOptions
* @property {BrazenJobRegistry} registry
* @property {object} kernel
* @property {() => number} [kernel.nextSeq]
* @property {AbortSignal} [kernel.abortSignal]
* @property {() => unknown} [kernel.readState]
* @property {(path: string) => unknown} [kernel.read]
* @property {boolean} [linkQueues]
*/
/**
* @typedef {object} JobRecord
* @property {string} type
* @property {string} key
* @property {unknown} payload
* @property {number} seq
* @property {number} priority
* @property {AbortSignal} abortSignal
*/
/** @type {Set<string>} */
const TERMINAL_JOB_STATUSES = new Set(['done', 'failed', 'skipped', 'cancelled', 'duplicate'])
/** Maximum retained terminal rows before pruning unreferenced keys. */
const TERMINAL_MAP_CAP = 512
/**
* @typedef {'done' | 'failed' | 'cancelled' | 'skipped' | 'duplicate'} JobTerminalStatus
*/
/**
* @param {string} type
* @param {string} key
* @return {string}
*/
function pendingMapKey(type, key)
{
return `${type}\0${key}`
}
/**
* @param {string} type
* @param {string} key
* @return {string}
*/
function terminalMapKey(type, key)
{
return `${type}:${key}`
}
/**
* @param {AbortSignal} parent
* @return {AbortController}
*/
function createScopedAbortController(parent)
{
let controller = new AbortController()
if (parent.aborted) {
controller.abort(parent.reason)
return controller
}
let onAbort = () => {
controller.abort(parent.reason)
}
parent.addEventListener('abort', onAbort, {once: true})
controller.signal.addEventListener('abort', () => {
parent.removeEventListener('abort', onAbort)
}, {once: true})
return controller
}
class BrazenScheduler
{
/**
* @param {SchedulerOptions} options
*/
constructor(options)
{
if (!options?.registry) {
throw new Error('BrazenScheduler: registry is required')
}
if (!options?.kernel) {
throw new Error('BrazenScheduler: kernel is required')
}
this._registry = options.registry
this._kernel = options.kernel
this._linkQueues = options.linkQueues === true
this._scopeController = createScopedAbortController(
options.kernel.abortSignal ?? new AbortController().signal,
)
/** @type {Map<string, JobRecord>} */
this._pending = new Map()
/** @type {Map<string, Promise<void>>} */
this._running = new Map()
/** @type {Map<string, JobTerminalStatus>} */
this._terminal = new Map()
/** @type {Map<string, AbortController>} */
this._jobControllers = new Map()
/** @type {Map<string, number>} */
this._runningCountByType = new Map()
this._localSeq = 0
this._idleWaiters = []
}
// -------------------------------------------------------------------------
// Public class methods
// -------------------------------------------------------------------------
/**
* Enqueue or coalesce a job; assigns seq from kernel when available.
* @param {string} type
* @param {unknown} payload
* @param {{ key?: string, priority?: number }} [options]
*/
schedule(type, payload, options = {})
{
let descriptor = this._registry.get(type)
if (!descriptor) {
throw new Error(`BrazenScheduler.schedule: unregistered job type "${type}"`)
}
let key = options.key
if (key == null) {
if (typeof descriptor.coalesceKey === 'function') {
key = descriptor.coalesceKey(payload)
} else {
key = `${type}:${JSON.stringify(payload)}`
}
}
key = String(key)
let mapKey = pendingMapKey(type, key)
let priority = options.priority ?? descriptor.priority ?? 0
let seq = this._nextSeq()
if (this._pending.has(mapKey)) {
let pending = this._pending.get(mapKey)
let incoming = {
type,
key,
payload,
seq,
priority: Math.max(pending.priority, priority),
abortSignal: pending.abortSignal,
}
let merge = descriptor.merge ?? ((_, next) => next)
let merged = merge(pending, incoming)
if (!merged) {
return
}
merged.priority = Math.max(pending.priority, priority)
merged.seq = Math.min(pending.seq, seq)
merged.abortSignal = pending.abortSignal
this._pending.set(mapKey, merged)
return
}
let controller = createScopedAbortController(this._scopeController.signal)
this._jobControllers.set(mapKey, controller)
this._pending.set(mapKey, {
type,
key,
payload,
seq,
priority,
abortSignal: controller.signal,
})
}
/** Spec alias for schedule. */
enqueue(type, payload, options = {})
{
this.schedule(type, payload, options)
}
/**
* Cancel pending work and abort in-flight execution for (type, key).
* @param {string} type
* @param {string} key
* @param {string} [reason]
*/
cancelPending(type, key, reason)
{
let mapKey = pendingMapKey(type, String(key))
let hadPending = this._pending.has(mapKey)
this._pending.delete(mapKey)
let controller = this._jobControllers.get(mapKey)
this._jobControllers.delete(mapKey)
if (controller && !controller.signal.aborted) {
controller.abort(reason ?? 'cancelled')
}
if (hadPending || this._running.has(mapKey)) {
this._markTerminal(type, String(key), 'cancelled')
}
}
/** Spec alias for cancelPending. */
cancel(type, key, reason)
{
this.cancelPending(type, key, reason)
}
/**
* Pump the ready set until idle or the scheduler scope is aborted.
* @return {Promise<void>}
*/
async runReady()
{
while (!this._isIdle() && !this._scopeController.signal.aborted) {
let started = this._startReadyJobs()
if (started === 0) {
if (this._running.size > 0) {
await this._waitForAnyRunning()
} else {
break
}
}
}
await this._waitForAllRunning()
}
/** Spec alias for runReady. */
async runUntilIdle()
{
return this.runReady()
}
/**
* @param {string} [type]
* @return {number}
*/
pendingCount(type)
{
if (!type) {
return this._pending.size
}
let count = 0
for (let job of this._pending.values()) {
if (job.type === type) {
count += 1
}
}
return count
}
// -------------------------------------------------------------------------
// Private class methods
// -------------------------------------------------------------------------
/**
* @return {number}
* @private
*/
_nextSeq()
{
if (typeof this._kernel.nextSeq === 'function') {
return this._kernel.nextSeq()
}
this._localSeq += 1
return this._localSeq
}
/**
* @return {boolean}
* @private
*/
_isIdle()
{
return this._pending.size === 0 && this._running.size === 0
}
/**
* @return {Promise<void>}
* @private
*/
_waitForAnyRunning()
{
if (this._running.size === 0) {
return Promise.resolve()
}
return Promise.race([...this._running.values()])
}
/**
* @return {Promise<void>}
* @private
*/
_waitForAllRunning()
{
if (this._running.size === 0) {
return Promise.resolve()
}
return Promise.all([...this._running.values()])
}
/**
* @private
*/
_notifyIdleWaiter()
{
if (this._isIdle() && this._idleWaiters.length) {
let waiters = this._idleWaiters.splice(0)
for (let resolve of waiters) {
resolve()
}
}
}
/**
* @param {string} type
* @param {string} key
* @param {JobTerminalStatus} status
* @private
*/
_markTerminal(type, key, status)
{
this._terminal.set(terminalMapKey(type, key), status)
this._pruneTerminalMap()
}
/**
* @private
*/
_pruneTerminalMap()
{
if (this._terminal.size <= TERMINAL_MAP_CAP) {
return
}
/** @type {Set<string>} */
let referenced = new Set()
for (let job of this._pending.values()) {
let descriptor = this._registry.get(job.type)
for (let depType of descriptor?.deps ?? []) {
referenced.add(terminalMapKey(depType, job.key))
}
}
for (let key of this._terminal.keys()) {
if (!referenced.has(key)) {
this._terminal.delete(key)
}
if (this._terminal.size <= TERMINAL_MAP_CAP) {
break
}
}
}
/**
* @param {JobRecord} job
* @return {boolean}
* @private
*/
_depsSatisfied(job)
{
let descriptor = this._registry.get(job.type)
let deps = descriptor?.deps ?? []
for (let depType of deps) {
let depTerminal = this._terminal.get(terminalMapKey(depType, job.key))
if (depTerminal && TERMINAL_JOB_STATUSES.has(depTerminal)) {
continue
}
if (this._pending.has(pendingMapKey(depType, job.key)) ||
this._running.has(pendingMapKey(depType, job.key))) {
return false
}
return false
}
return true
}
/**
* @param {string} type
* @return {number}
* @private
*/
_concurrencyLimit(type)
{
let descriptor = this._registry.get(type)
return descriptor?.concurrency ?? 1
}
/**
* @param {string} type
* @return {number}
* @private
*/
_runningCountForType(type)
{
return this._runningCountByType.get(type) ?? 0
}
/**
* @return {JobRecord[]}
* @private
*/
_collectReadyJobs()
{
/** @type {JobRecord[]} */
let ready = []
for (let job of this._pending.values()) {
if (!this._depsSatisfied(job)) {
continue
}
if (this._runningCountForType(job.type) >= this._concurrencyLimit(job.type)) {
continue
}
ready.push(job)
}
ready.sort((a, b) => {
if (this._linkQueues) {
let rank = (type) => {
if (type === 'resolve') {
return 0
}
if (type === 'download') {
return 1
}
return 2
}
let rankDiff = rank(a.type) - rank(b.type)
if (rankDiff !== 0) {
return rankDiff
}
}
if (b.priority !== a.priority) {
return b.priority - a.priority
}
return a.seq - b.seq
})
/** @type {JobRecord[]} */
let selected = []
/** @type {Map<string, number>} */
let selectedCounts = new Map()
for (let job of ready) {
let count = selectedCounts.get(job.type) ?? 0
if (count >= this._concurrencyLimit(job.type)) {
continue
}
selected.push(job)
selectedCounts.set(job.type, count + 1)
}
return selected
}
/**
* @return {number}
* @private
*/
_startReadyJobs()
{
let ready = this._collectReadyJobs()
for (let job of ready) {
let mapKey = pendingMapKey(job.type, job.key)
this._pending.delete(mapKey)
let runPromise = this._executeJob(job)
this._running.set(mapKey, runPromise)
this._runningCountByType.set(job.type, this._runningCountForType(job.type) + 1)
runPromise.finally(() => {
this._running.delete(mapKey)
this._runningCountByType.set(
job.type,
Math.max(0, this._runningCountForType(job.type) - 1),
)
this._jobControllers.delete(mapKey)
this._notifyIdleWaiter()
})
}
return ready.length
}
/**
* @param {JobRecord} job
* @return {Promise<void>}
* @private
*/
async _executeJob(job)
{
let descriptor = this._registry.get(job.type)
if (!descriptor) {
this._markTerminal(job.type, job.key, 'failed')
return
}
if (job.abortSignal.aborted) {
this._markTerminal(job.type, job.key, 'cancelled')
return
}
let readState = () => {
if (typeof this._kernel.readState === 'function') {
return this._kernel.readState()
}
return {}
}
if (typeof descriptor.reassess === 'function') {
let reassessed = descriptor.reassess(job, readState)
if (reassessed == null) {
this._markTerminal(job.type, job.key, 'skipped')
return
}
job = {...job, ...reassessed, abortSignal: job.abortSignal}
}
/** @type {import('./reactor-core.spec.md').JobContext} */
let ctx = {
kernel: this._kernel,
repos: this._kernel.repos ?? {},
abortSignal: job.abortSignal,
spawn: (type, spawnPayload) => {
this.schedule(type, spawnPayload, {key: job.key})
},
read: (path) => {
if (typeof this._kernel.read === 'function') {
return this._kernel.read(path)
}
let state = readState()
if (state && typeof state === 'object' && path in state) {
return state[path]
}
return undefined
},
yield: () => new Promise((resolve) => setTimeout(resolve, 0)),
}
try {
await descriptor.run(ctx, job.payload)
if (job.abortSignal.aborted) {
this._markTerminal(job.type, job.key, 'cancelled')
} else {
this._markTerminal(job.type, job.key, 'done')
}
} catch (error) {
if (job.abortSignal.aborted) {
this._markTerminal(job.type, job.key, 'cancelled')
} else {
this._markTerminal(job.type, job.key, 'failed')
}
}
}
}
globalThis.BrazenScheduler = BrazenScheduler
// -------------------------------------------------------------------------
// Kernel
// -------------------------------------------------------------------------
const KERNEL_SEQ_MAX_SAFE = Number.MAX_SAFE_INTEGER - 1
/**
* @param {string} scriptPrefix
* @return {string}
*/
function coordinatorLockName(scriptPrefix)
{
return `brazen-${String(scriptPrefix ?? '').trim()}-coordinator`
}
class BrazenKernel
{
// -------------------------------------------------------------------------
// Static public methods
// -------------------------------------------------------------------------
/**
* @param {string} scriptPrefix
* @return {string}
*/
static coordinatorLockName(scriptPrefix)
{
return coordinatorLockName(scriptPrefix)
}
// -------------------------------------------------------------------------
// Protected class variables
// -------------------------------------------------------------------------
/** @type {string} */
_scriptPrefix = ''
/** @type {string} */
_lockName = ''
/** @type {object|null} */
_signals = null
/** @type {object|null} */
_bus = null
/** @type {object|null} */
_repos = null
/** @type {object|null} */
_scheduler = null
/** @type {(() => void)|null} */
_onCoordinatorAcquired = null
/** @type {((reason: 'steal'|'release'|'abort') => void)|null} */
_onCoordinatorLost = null
/** @type {((command: object, seq: number, patches: object[]) => void)|null} */
_onCommandApplied = null
/** @type {AbortController} */
_abortController = new AbortController()
/** @type {number} */
_seq = 0
/** @type {boolean} */
_isCoordinator = false
/** @type {boolean} */
_started = false
/** @type {boolean} */
_hydrated = false
/** @type {boolean} */
_documentSuspended = false
/** @type {(() => void)|null} */
_releaseLock = null
/** @type {'steal'|'release'|'abort'|null} */
_lossReason = null
/** @type {Promise<void>|null} */
_lockLoopPromise = null
/** @type {(() => void)|null} */
_busUnsubscribe = null
/** @type {boolean} */
_lifecycleBound = false
/** @type {Map<string, unknown>} */
_authoritativeValues = new Map()
/** @type {{seq: number, patches: object[]}[]} */
_committedPatches = []
/** @type {Promise<void>|null} */
_hydratePromise = null
// -------------------------------------------------------------------------
// Constructor
// -------------------------------------------------------------------------
/**
* @param {object} options
* @param {string} options.scriptPrefix
* @param {object} options.signals
* @param {object} options.bus
* @param {object} [options.repos]
* @param {object} options.scheduler
* @param {() => void} [options.onCoordinatorAcquired]
* @param {(reason: 'steal'|'release'|'abort') => void} [options.onCoordinatorLost]
* @param {(command: object, seq: number, patches: object[]) => void} [options.onCommandApplied]
*/
constructor(options = {})
{
this._scriptPrefix = String(options.scriptPrefix ?? '').trim()
this._lockName = coordinatorLockName(this._scriptPrefix)
this._signals = options.signals ?? null
this._bus = options.bus ?? null
this._repos = options.repos ?? null
this._scheduler = options.scheduler ?? null
this._onCoordinatorAcquired = typeof options.onCoordinatorAcquired === 'function'
? options.onCoordinatorAcquired
: null
this._onCoordinatorLost = typeof options.onCoordinatorLost === 'function'
? options.onCoordinatorLost
: null
this._onCommandApplied = typeof options.onCommandApplied === 'function'
? options.onCommandApplied
: null
}
// -------------------------------------------------------------------------
// Public getters
// -------------------------------------------------------------------------
/** @return {AbortSignal} */
get abortSignal()
{
return this._abortController.signal
}
// -------------------------------------------------------------------------
// Public class methods
// -------------------------------------------------------------------------
/**
* Start lock acquisition loop + bus command handler. Idempotent.
* @return {Promise<void>}
*/
async start()
{
if (this._started) {
return
}
this._started = true
this._documentSuspended = false
this._bindLifecycle()
this._ensureBusSubscription()
this._lockLoopPromise = this._runCoordinatorLockLoop()
}
/**
* Release lock gracefully (pagehide). Aborts in-flight jobs.
* @param {'pagehide'|'manual'} [reason]
* @return {Promise<void>}
*/
async stop(reason = 'manual')
{
if (reason === 'pagehide') {
this._documentSuspended = true
this._lossReason = 'abort'
} else {
this._lossReason = 'release'
}
this._started = false
if (this._releaseLock) {
let release = this._releaseLock
this._releaseLock = null
release()
}
if (this._lockLoopPromise) {
try {
await this._lockLoopPromise
} catch (e) {
// lock loop may reject after steal
}
this._lockLoopPromise = null
}
}
/**
* @return {boolean}
*/
isCoordinator()
{
return this._isCoordinator
}
/**
* @param {{steal?: boolean, ifAvailable?: boolean}} [options]
* @return {Promise<boolean>}
*/
async requestCoordinatorRole(options = {})
{
if (this._documentSuspended || !this._locksAvailable()) {
return false
}
if (this.isCoordinator()) {
return true
}
let steal = options.steal === true
let ifAvailable = options.ifAvailable === true
if (steal) {
return this._requestLock({steal: true})
}
if (ifAvailable) {
return this._requestLock({ifAvailable: true})
}
return this._requestLock({ifAvailable: true})
}
/**
* @return {number}
*/
getSeq()
{
return this._seq
}
/**
* @param {object} command
* @return {Promise<void>}
*/
async dispatch(command)
{
if (!this.isCoordinator()) {
throw new Error('BrazenKernel.dispatch requires coordinator role')
}
await this.handleCommand({
kind: 'command',
tabId: this._bus?.tabId ?? 'local',
command,
})
}
/**
* Coordinator: emit patches after atom mutation + write-through (SW-1 bridge for dm.state.*).
* @param {object} command stub command routed to repos.kernelWriteThrough (SW-2 expands handlers)
* @param {object[]} patches `{path, value}` entries; seq/version assigned here
* @return {Promise<void>}
*/
async commitStatePatches(command, patches)
{
if (!this.isCoordinator() || !patches?.length) {
return
}
if (!this._hydrated) {
await this.hydrate()
}
let seq = this._assignSeq()
let stamped = this._stampPatchesForCommit(patches, seq)
await this._writeThrough(command, stamped)
this._recordCommittedPatch(seq, stamped)
this._publishPatch(seq, stamped)
}
/**
* Hydrate atoms from IDB + serve snapshot responses.
* @return {Promise<void>}
*/
async hydrate()
{
if (this._hydratePromise) {
return this._hydratePromise
}
this._hydratePromise = this._doHydrate()
try {
await this._hydratePromise
} finally {
this._hydratePromise = null
}
}
/**
* Apply command: validate → mutate atoms → schedule jobs → IDB write-through → broadcast patch.
* @param {object} message
* @return {Promise<void>}
*/
async handleCommand(message)
{
if (!this.isCoordinator() || !message || message.kind !== 'command') {
return
}
let command = message.command
if (!command || typeof command.type !== 'string') {
return
}
if (command.type === 'request-coordinator-steal') {
await this.requestCoordinatorRole({steal: true})
return
}
if (!this._hydrated) {
await this.hydrate()
}
let seq = this._assignSeq()
let patches = this._mutateAtomsForCommand(command, seq)
await this._writeThrough(command, patches)
this._scheduleJobsForCommand(command, seq)
this._recordCommittedPatch(seq, patches)
this._publishPatch(seq, patches)
this._onCommandApplied?.(command, seq, patches)
}
/**
* Coordinator: respond to snapshot-request.
* @param {object} request
* @return {Promise<void>}
*/
async handleSnapshotRequest(request)
{
if (!this.isCoordinator() || !request || request.kind !== 'snapshot-request') {
return
}
if (!this._hydrated) {
await this.hydrate()
}
let sinceSeq = Number(request.sinceSeq) || 0
let snapshot = this._buildSnapshotRecord()
let catchUpPatches = sinceSeq > 0
? this._flattenCatchUpPatches(sinceSeq, this._seq)
: undefined
if (typeof this._bus?.publish === 'function') {
this._bus.publish({
kind: 'snapshot-response',
tabId: this._bus.tabId,
requestId: request.requestId,
snapshot,
snapshotSeq: this._seq,
catchUpPatches,
})
}
}
// -------------------------------------------------------------------------
// Private class methods
// -------------------------------------------------------------------------
/**
* @return {Promise<void>}
* @private
*/
async _runCoordinatorLockLoop()
{
while (this._started && !this._documentSuspended) {
try {
await this._holdCoordinatorLock({})
} catch (e) {
if (this._isCoordinator) {
this._loseCoordinator('steal')
}
}
if (!this._started) {
break
}
}
}
/**
* @param {{steal?: boolean, ifAvailable?: boolean}} options
* @return {Promise<boolean>}
* @private
*/
async _requestLock(options = {})
{
if (!this._locksAvailable()) {
return false
}
return new Promise((resolveClaim) => {
void this._holdCoordinatorLock(options, resolveClaim)
})
}
/**
* @param {{steal?: boolean, ifAvailable?: boolean}} options
* @param {(claimed: boolean) => void} [onClaimed]
* @return {Promise<void>}
* @private
*/
async _holdCoordinatorLock(options = {}, onClaimed = null)
{
if (!this._locksAvailable()) {
onClaimed?.(false)
return
}
let lockOptions = {mode: 'exclusive'}
if (options.steal) {
lockOptions.steal = true
}
if (options.ifAvailable) {
lockOptions.ifAvailable = true
}
let claimedCalled = false
/** @param {boolean} claimed */
let claimOnce = (claimed) => {
if (claimedCalled) {
return
}
claimedCalled = true
onClaimed?.(claimed)
}
try {
await navigator.locks.request(this._lockName, lockOptions, async (lock) => {
if (!lock) {
claimOnce(false)
return
}
claimOnce(true)
this._becomeCoordinator()
try {
await new Promise((resolve) => {
this._releaseLock = resolve
})
} finally {
this._loseCoordinator(this._lossReason ?? 'release')
this._lossReason = null
this._releaseLock = null
}
})
} catch (error) {
if (this._isCoordinator) {
this._loseCoordinator('steal')
} else {
claimOnce(false)
}
throw error
}
}
/**
* @private
*/
_becomeCoordinator()
{
if (this._isCoordinator) {
return
}
this._abortController = new AbortController()
this._isCoordinator = true
if (!this._hydrated) {
void this.hydrate()
}
this._onCoordinatorAcquired?.()
}
/**
* @param {'steal'|'release'|'abort'} reason
* @private
*/
_loseCoordinator(reason)
{
if (!this._isCoordinator) {
return
}
this._isCoordinator = false
this._abortController.abort()
if (typeof this._scheduler?.cancelAll === 'function') {
this._scheduler.cancelAll(this.abortSignal.reason ?? reason)
}
this._onCoordinatorLost?.(reason)
}
/**
* @return {boolean}
* @private
*/
_locksAvailable()
{
return typeof navigator !== 'undefined' &&
navigator.locks != null &&
typeof navigator.locks.request === 'function'
}
/**
* @private
*/
_bindLifecycle()
{
if (this._lifecycleBound || typeof document === 'undefined') {
return
}
this._lifecycleBound = true
document.addEventListener('pagehide', (event) => {
void this.stop('pagehide')
})
document.addEventListener('pageshow', (event) => {
if (event.persisted) {
this._documentSuspended = false
void this.start()
}
})
}
/**
* @private
*/
_ensureBusSubscription()
{
if (this._busUnsubscribe || typeof this._bus?.subscribe !== 'function') {
return
}
this._busUnsubscribe = this._bus.subscribe((message) => {
if (message?.kind === 'command') {
void this.handleCommand(message)
} else if (message?.kind === 'snapshot-request') {
void this.handleSnapshotRequest(message)
}
})
}
/**
* @return {Promise<void>}
* @private
*/
async _doHydrate()
{
if (typeof this._signals?.batch === 'function') {
this._signals.batch(() => {
this._hydrateAuthoritativePathsFromRepos()
})
} else {
this._hydrateAuthoritativePathsFromRepos()
}
this._hydrated = true
}
/**
* @private
*/
_hydrateAuthoritativePathsFromRepos()
{
if (!this._repos) {
return
}
// Phase 1 stub: repos hook for Phase 2 IDB hydration.
if (typeof this._repos.kernelHydrate === 'function') {
let snapshot = this._repos.kernelHydrate()
if (snapshot && typeof snapshot === 'object') {
for (let [path, value] of Object.entries(snapshot)) {
this._authoritativeValues.set(path, structuredClone(value))
this._writeSignalPath(path, value, 0)
}
}
}
}
/**
* @param {object} command
* @param {number} seq
* @return {object[]}
* @private
*/
_stampPatchesForCommit(patches, seq)
{
let stamped = []
let writeOne = (path, value) => {
let cloned = structuredClone(value)
this._authoritativeValues.set(path, cloned)
let version = seq
if (typeof this._signals?.batch === 'function') {
this._signals.batch(() => {
version = this._writeSignalPath(path, cloned, seq) ?? seq
})
} else {
version = this._writeSignalPath(path, cloned, seq) ?? seq
}
stamped.push({path, version, value: cloned})
}
for (let patch of patches) {
if (!patch?.path) {
continue
}
writeOne(patch.path, patch.value)
}
return stamped
}
/**
* @param {object} command
* @param {number} seq
* @return {object[]}
* @private
*/
_mutateAtomsForCommand(command, seq)
{
let patches = []
let apply = (path, value) => {
let version = seq
this._authoritativeValues.set(path, structuredClone(value))
if (typeof this._signals?.batch === 'function') {
this._signals.batch(() => {
version = this._writeSignalPath(path, value, seq) ?? seq
})
} else {
version = this._writeSignalPath(path, value, seq) ?? seq
}
patches.push({path, version, value: structuredClone(value)})
}
switch (command.type) {
case 'toggle-paused':
apply('dm.state.paused', Boolean(command.payload?.paused))
break
case 'write-setting':
apply(`config.settings.${command.payload?.fieldKey}`, command.payload?.value)
break
case 'enqueue-download':
case 'dequeue-download':
case 'confirm-tag-discovery':
case 'skip-tag-discovery':
case 'clear-download-queue':
apply('kernel.lastCommand', {type: command.type, payload: command.payload ?? {}})
break
case 'config-save':
case 'config-sync':
apply('kernel.lastCommand', {type: command.type, payload: command.payload ?? {}})
break
case 'custom':
if (command.payload?.name !== 'pipeline-pump') {
apply(`kernel.custom.${command.payload?.name ?? 'unknown'}`, command.payload?.data)
}
break
default:
apply(`kernel.lastCommand`, {type: command.type, payload: command.payload ?? {}})
break
}
return patches
}
/**
* @param {string} path
* @param {unknown} value
* @param {number} fallbackVersion
* @return {number|undefined}
* @private
*/
_writeSignalPath(path, value, fallbackVersion)
{
if (typeof this._signals?.atom !== 'function') {
return fallbackVersion
}
let atomRef = this._signals.atom(path, value)
if (atomRef && typeof atomRef.write === 'function') {
return atomRef.write(value)
}
return fallbackVersion
}
/**
* @param {object} command
* @param {object[]} patches
* @return {Promise<void>}
* @private
*/
async _writeThrough(command, patches)
{
if (typeof this._repos?.kernelWriteThrough === 'function') {
await this._repos.kernelWriteThrough(command, patches, this.getSeq())
}
}
/**
* @param {object} command
* @param {number} seq
* @private
*/
_scheduleJobsForCommand(command, seq)
{
if (typeof this._scheduler?.enqueue !== 'function') {
return
}
if (command.type === 'spawn-job') {
this._scheduler.enqueue(
command.payload?.jobType ?? 'unknown',
command.payload?.payload,
{key: command.payload?.key, priority: seq},
)
return
}
if (command.type === 'enqueue-download') {
this._scheduler.enqueue('resolve', command.payload, {key: command.payload?.itemId, priority: seq})
return
}
if (command.type === 'config-sync') {
this._scheduler.enqueue('config-sync', command.payload, {
key: command.payload?.coalesceKey ?? 'config',
priority: seq,
})
}
}
/**
* @param {number} seq
* @param {object[]} patches
* @private
*/
_recordCommittedPatch(seq, patches)
{
this._committedPatches.push({seq, patches: structuredClone(patches)})
if (this._committedPatches.length > 512) {
this._committedPatches.splice(0, this._committedPatches.length - 512)
}
}
/**
* @param {number} seq
* @param {object[]} patches
* @private
*/
_publishPatch(seq, patches)
{
if (typeof this._bus?.publish !== 'function') {
return
}
this._bus.publish({
kind: 'patch',
tabId: this._bus.tabId,
seq,
ts: Date.now(),
patches: structuredClone(patches),
})
}
/**
* @return {number}
* @private
*/
_assignSeq()
{
let next
if (typeof this._bus?.nextSeq === 'function') {
next = this._bus.nextSeq()
} else {
next = this._seq + 1
}
if (!Number.isFinite(next) || next <= this._seq) {
next = this._seq + 1
}
if (next > KERNEL_SEQ_MAX_SAFE) {
throw new Error('BrazenKernel seq overflow')
}
this._seq = next
return next
}
/**
* @return {Record<string, unknown>}
* @private
*/
_buildSnapshotRecord()
{
let snapshot = {}
for (let [path, value] of this._authoritativeValues.entries()) {
snapshot[path] = structuredClone(value)
}
return snapshot
}
/**
* @param {number} sinceSeq
* @param {number} snapshotSeq
* @return {object[]|undefined}
* @private
*/
_flattenCatchUpPatches(sinceSeq, snapshotSeq)
{
let out = []
for (let entry of this._committedPatches) {
if (entry.seq > sinceSeq && entry.seq <= snapshotSeq) {
out.push(...entry.patches.map((patch) => structuredClone(patch)))
}
}
return out.length ? out : undefined
}
}
globalThis.BrazenKernel = BrazenKernel
globalThis.coordinatorLockName = coordinatorLockName