Reactor Core: signals, event bus, job scheduler, and coordinator kernel
此腳本不應該直接安裝，它是一個供其他腳本使用的函式庫。欲使用本函式庫，請在腳本 metadata 寫上： // @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