novy-syndicate/src/actor.ts

import { Dictionary, IdentitySet, Record, Tuple, Value, is, IdentityMap } from 'preserves';
import { queueTask } from './task.js';

//---------------------------------------------------------------------------

if ('stackTraceLimit' in Error) {
    Error.stackTraceLimit = Infinity;
}

export type Assertion = Value<Ref>;
export type Handle = number;
export type ExitReason = null | { ok: true } | { ok: false, err: Error };
export type LocalAction = (t: Turn) => void;

export interface Entity {
    assert(turn: Turn, assertion: Assertion, handle: Handle): void;
    retract(turn: Turn, handle: Handle): void;
    message(turn: Turn, body: Assertion): void;
    sync(turn: Turn, peer: Ref): void;
}

export interface Ref {
    readonly relay: Actor;
    readonly target: Partial<Entity>;
    readonly attenuation?: Attenuation;
}

export type Attenuation = Array<RewriteStage>; // array of stages, each a list of alternatives
export type RewriteStage = Array<Rewrite>;
export type Rewrite = { pattern: Pattern, template: Template };

export const _CRec = Symbol.for('rec');
export const CRec = Record.makeConstructor<{label: Assertion, arity: number}, Ref>()(
    _CRec, ['label', 'arity']);

export const _CArr = Symbol.for('arr');
export const CArr = Record.makeConstructor<{arity: number}, Ref>()(
    _CArr, ['arity']);

export const _CDict = Symbol.for('dict');
export const CDict = Record.makeConstructor<{}, Ref>()(
    _CDict, []);

export type ConstructorSpec =
    | Record<typeof _CRec, [Assertion, number], Ref>
    | Record<typeof _CArr, [number], Ref>
    | Record<typeof _CDict, [], Ref>;

export const _PDiscard = Symbol.for('_');
export const PDiscard = Record.makeConstructor<{}, Ref>()(
    _PDiscard, []);

export const _PBind = Symbol.for('bind');
export const PBind = Record.makeConstructor<{name: string, pattern: Pattern}, Ref>()(
    _PBind, ['name', 'pattern']);

export const _PAnd = Symbol.for('and');
export const PAnd = Record.makeConstructor<{patterns: Array<Pattern>}, Ref>()(
    _PAnd, ['patterns']);

export const _PNot = Symbol.for('not');
export const PNot = Record.makeConstructor<{pattern: Pattern}, Ref>()(
    _PNot, ['pattern']);

export const _Lit = Symbol.for('lit');
export const Lit = Record.makeConstructor<{value: Assertion}, Ref>()(
    _Lit, ['value']);

export const _PCompound = Symbol.for('compound');
export const PCompound =
    Record.makeConstructor<{ctor: ConstructorSpec, members: Dictionary<Pattern, Ref>}, Ref>()(
        _PCompound, ['ctor', 'members']);

export type Pattern =
    | Record<typeof _PDiscard, [], Ref>
    | Record<typeof _PBind, [string, Pattern], Ref>
    | Record<typeof _PAnd, [Pattern[]], Ref>
    | Record<typeof _PNot, [Pattern], Ref>
    | Record<typeof _Lit, [Assertion], Ref>
    | Record<typeof _PCompound, [ConstructorSpec, Dictionary<Pattern, Ref>], Ref>;

export const _TRef = Symbol.for('ref');
export const TRef = Record.makeConstructor<{name: string}, Ref>()(
    _TRef, ['name']);

export const _TCompound = Symbol.for('compound');
export const TCompound =
    Record.makeConstructor<{ctor: ConstructorSpec, members: Dictionary<Template, Ref>}, Ref>()(
        _TCompound, ['ctor', 'members']);

export type Template =
    | Record<typeof _TRef, [string], Ref>
    | Record<typeof _Lit, [Assertion], Ref>
    | Record<typeof _TCompound, [ConstructorSpec, Dictionary<Template, Ref>], Ref>;

export type Bindings = { [name: string]: Assertion };

//---------------------------------------------------------------------------

export function isRef(v: any): v is Ref {
    return 'relay' in v && v.relay instanceof Actor && 'target' in v;
}

let nextActorId = 0;
export const __setNextActorId = (v: number) => nextActorId = v;

export class Actor {
    readonly id = nextActorId++;
    readonly outbound: Map<Handle, Ref>;
    exitReason: ExitReason = null;
    // readonly exitHooks: Array<LocalAction> = [];

    constructor(initialAssertions = new Map<Handle, Ref>()) {
        this.outbound = initialAssertions;
    }

    // atExit(a: LocalAction): void {
    //     this.exitHooks.push(a);
    // }

    terminateWith(t: Turn, reason: Exclude<ExitReason, null>) {
        if (this.exitReason !== null) return;
        this.exitReason = reason;
        if (!this.exitReason.ok) {
            console.error(`Actor ${this.id} crashed:`, this.exitReason.err);
        }
        // this.exitHooks.forEach(hook => hook(t));
        queueTask(() =>
            t.freshen(t =>
                this.outbound.forEach((peer, h) => t._retract(peer, h))));
    }

    execute(proc: () => void): void {
        queueTask(() => {
            if (this.exitReason !== null) return;
            try {
                proc();
            } catch (err) {
                Turn.for(this, t => this.terminateWith(t, { ok: false, err }));
            }
        });
    }
}

let nextHandle = 0;
function allocateHandle(): Handle {
    return nextHandle++;
}

export function _sync_impl(turn: Turn, e: Partial<Entity>, peer: Ref): void {
    e.sync ? e.sync!(turn, peer) : turn.message(peer, true);
}

let nextTurnId = 0;

export class Turn {
    readonly id = nextTurnId++;
    readonly actor: Actor;
    queues: Map<Actor, LocalAction[]> | null = new Map();

    static for(actor: Actor, f: LocalAction): void {
        const t = new Turn(actor);
        f(t);
        t.queues!.forEach((q, a) => a.execute(() => q.forEach(f => Turn.for(a, f))));
        t.queues = null;
    }

    private constructor(actor: Actor) {
        this.actor = actor;
    }

    ref<T extends Partial<Entity>>(e: T): Ref {
        return { relay: this.actor, target: e };
    }

    spawn(bootProc: LocalAction, initialAssertions = new IdentitySet<Handle>()): void {
        this.enqueue(this.actor, () => {
            const newOutbound = new Map<Handle, Ref>();
            initialAssertions.forEach(key => {
                newOutbound.set(key, this.actor.outbound.get(key)!); // we trust initialAssertions
                this.actor.outbound.delete(key);
            });
            const child = new Actor(newOutbound);
            child.execute(() => Turn.for(child, bootProc));
        });
    }

    quit(): void {
        this.enqueue(this.actor, t => this.actor.terminateWith(t, { ok: true }));
    }

    crash(err: Error): void {
        this.enqueue(this.actor, t => this.actor.terminateWith(t, { ok: false, err }));
    }

    assert(ref: Ref, assertion: Assertion): Handle {
        const h = allocateHandle();
        this._assert(ref, assertion, h);
        return h;
    }

    _assert(ref: Ref, assertion: Assertion, h: Handle) {
        const a = runRewrites(ref.attenuation, assertion);
        if (a !== null) {
            this.enqueue(ref.relay, t => {
                this.actor.outbound.set(h, ref);
                ref.target.assert?.(t, a, h);
            });
        }
    }

    retract(h: Handle): void {
        const peer = this.actor.outbound.get(h);
        if (peer === void 0) return;
        this._retract(peer, h);
    }

    replace(ref: Ref, h: Handle | undefined, assertion: Assertion): Handle {
        const newHandle = this.assert(ref, assertion);
        if (h !== void 0) this.retract(h);
        return newHandle;
    }

    _retract(ref: Ref, handle: Handle): void {
        this.enqueue(ref.relay, t => {
            this.actor.outbound.delete(handle);
            ref.target.retract?.(t, handle);
        });
    }

    sync(ref: Ref): Promise<Turn> {
        return new Promise(resolve => this._sync(ref, this.ref({ message: resolve })));
    }

    _sync(ref: Ref, peer: Ref): void {
        this.enqueue(ref.relay, t => _sync_impl(t, ref.target, peer));
    }

    message(ref: Ref, assertion: Assertion): void {
        const a = runRewrites(ref.attenuation, assertion);
        if (a !== null) this.enqueue(ref.relay, t => ref.target.message?.(t, assertion));
    }

    enqueue(relay: Actor, a: LocalAction): void {
        if (this.queues === null) {
            throw new Error("Attempt to reuse a committed Turn");
        }
        this.queues.get(relay)?.push(a) ?? this.queues.set(relay, [a]);
    }

    freshen(a: LocalAction): void {
        if (this.queues !== null) {
            throw new Error("Attempt to freshen a non-stale Turn");
        }
        Turn.for(this.actor, a);
    }
}

//---------------------------------------------------------------------------

export function match(p: Pattern, v: Assertion): Bindings | null {
    let bindings: Bindings = {};

    function walk(p: Pattern, v: Assertion): boolean {
        switch (p.label) {
            case _PDiscard:
                return true;
            case _PBind:
                if (walk(PBind._.pattern(p), v)) {
                    bindings[PBind._.name(p)] = v;
                    return true;
                }
                return false;
            case _PAnd:
                for (const pp of PAnd._.patterns(p)) {
                    if (!walk(pp, v)) return false;
                }
                return true;
            case _PNot: {
                const savedBindings = bindings;
                bindings = {};
                const result = !walk(PNot._.pattern(p), v)
                bindings = savedBindings;
                return result;
            }
            case _Lit:
                return is(Lit._.value(p), v);
            case _PCompound: {
                const ctor = PCompound._.ctor(p);
                const members = PCompound._.members(p);
                switch (ctor.label) {
                    case _CRec:
                        if (!Record.isRecord<Assertion, Tuple<Assertion>, Ref>(v)) return false;
                        if (!is(CRec._.label(ctor), v.label)) return false;
                        if (CRec._.arity(ctor) !== v.length) return false;
                        for (const [key, pp] of members) {
                            if (typeof key !== 'number') return false;
                            if (!walk(pp, v[key])) return false;
                        }
                        return true;
                    case _CArr:
                        if (!Array.isArray(v)) return false;
                        if ('label' in v) return false;
                        if (CArr._.arity(ctor) !== v.length) return false;
                        for (const [key, pp] of members) {
                            if (typeof key !== 'number') return false;
                            if (!walk(pp, v[key])) return false;
                        }
                        return true;
                    case _CDict:
                        if (!Dictionary.isDictionary<Assertion, Ref>(v)) return false;
                        for (const [key, pp] of members) {
                            const vv = v.get(key);
                            if (vv === void 0) return false;
                            if (!walk(pp, vv)) return false;
                        }
                        return true;
                }
            }
        }
    }

    return walk(p, v) ? bindings : null;
}

export function instantiate(t: Template, b: Bindings): Assertion {
    function walk(t: Template): Assertion {
        switch (t.label) {
            case _TRef: {
                const v = b[TRef._.name(t)];
                if (v === void 0) throw new Error(`Unbound reference: ${TRef._.name(t)}`);
                return v;
            }
            case _Lit:
                return Lit._.value(t);
            case _TCompound: {
                const ctor = TCompound._.ctor(t);
                const members = TCompound._.members(t);
                switch (ctor.label) {
                    case _CRec: {
                        const v = Record<Assertion, any, Ref>(CRec._.label(ctor), []);
                        v.length = CRec._.arity(ctor);
                        for (const [key, tt] of members) {
                            v[key as number] = walk(tt);
                        }
                        return v;
                    }
                    case _CArr: {
                        const v = [];
                        v.length = CArr._.arity(ctor);
                        for (const [key, tt] of members) {
                            v[key as number] = walk(tt);
                        }
                        return v;
                    }
                    case _CDict: {
                        const v = new Dictionary<Assertion, Ref>();
                        for (const [key, tt] of members) {
                            v.set(key, walk(tt));
                        }
                        return v;
                    }
                }
            }
        }
    }

    return walk(t);
}

export function rewrite(r: Rewrite, v: Assertion): Assertion | null {
    const bindings = match(r.pattern, v);
    if (bindings === null) return null;
    return instantiate(r.template, bindings);
}

export function examineAlternatives(alternatives: RewriteStage, v: Assertion): Assertion | null {
    for (const r of alternatives) {
        const w = rewrite(r, v);
        if (w !== null) return w;
    }
    return null;
}

export function runRewrites(a: Attenuation | undefined, v: Assertion): Assertion | null {
    if (a !== void 0) {
        for (const stage of a) {
            const w = examineAlternatives(stage, v);
            if (w === null) return null;
            v = w;
        }
    }
    return v;
}

export function rfilter(... patterns: Pattern[]): RewriteStage {
    return patterns.map(p => ({ pattern: PBind('a', p), template: TRef('a') }));
}

export function attenuate(ref: Ref, ... a: Attenuation): Ref {
    return { ... ref, attenuation: [... a, ... (ref.attenuation ?? [])] };
}

export function forwarder(t: Turn, ref: Ref): { proxy: Ref, revoker: Ref } {
    let underlying: Ref | null = ref;
    let handleMap = new IdentityMap<Handle, Handle>();
    let proxy = t.ref({
        assert(turn: Turn, assertion: Assertion, handle: Handle): void {
            if (underlying === null) return;
            handleMap.set(handle, turn.assert(underlying, assertion));
        },
        retract(turn: Turn, handle: Handle): void {
            if (underlying === null) return;
            const h = handleMap.get(handle);
            if (h !== void 0) {
                turn.retract(h);
                handleMap.delete(handle);
            }
        },
        message(turn: Turn, body: Assertion): void {
            if (underlying === null) return;
            turn.message(underlying, body);
        },
        sync(turn: Turn, peer: Ref): void {
            if (underlying === null) return;
            turn._sync(underlying, peer);
        },
    });
    let revoker = t.ref({
        message(turn: Turn, _body: Assertion): void {
            underlying = null;
            handleMap.forEach(h => turn.retract(h));
        },
    });
    return { proxy, revoker };
}