novy-syndicate/actor.ts

import {
    Dictionary,
    IdentitySet,
    Record,
    Value,
    fold,
    is,
    mapPointers,
    unannotate,
} from 'preserves';

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

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<Rewrite>;
export type Rewrite = { pattern: Pattern, template: Template };

export type PatternPointer =
    | Ref
    | { type: 'discard' }
    | { type: 'bind', name: string, pattern: Pattern }
    | { type: 'and', patterns: Array<Pattern> }
    | { type: 'or', patterns: Array<Pattern> }
    | { type: 'not', pattern: Pattern };
export type Pattern = Value<PatternPointer>;

export type TemplatePointer = Ref | { type: 'ref', name: string };
export type Template = Value<TemplatePointer>;

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;
}

export const pdiscard: Pattern = { type: 'discard' };
export const pbind = (name: string, pattern: Pattern = pdiscard): Pattern =>
    ({ type: 'bind', name, pattern });
export const pand = (... patterns: Pattern[]): Pattern => ({ type: 'and', patterns });
export const por = (... patterns: Pattern[]): Pattern => ({ type: 'or', patterns });
export const pnot = (pattern: Pattern): Pattern => ({ type: 'not', pattern });

export const tref = (name: string): Template => ({ type: 'ref', name });

export class Actor {
    readonly outbound: Map<Handle, Ref>;
    exitReason: ExitReason = null;

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

    terminateWith(t: Turn, reason: Exclude<ExitReason, null>) {
        if (this.exitReason !== null) return;
        this.exitReason = reason;
        this.outbound.forEach((peer, h) => t._retract(peer, h));
    }

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

let nextHandle = 0;

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

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

    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))));
    }

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

    ref(e: Partial<Entity>): 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 }));
    }

    assert(ref: Ref, assertion: Assertion): Handle {
        const h = nextHandle++;
        this.enqueue(ref.relay, t => {
            const a = ref.attenuation === void 0
                ? assertion
                : runRewrites(ref.attenuation, assertion);
            if (a !== null) {
                this.actor.outbound.set(h, ref);
                ref.target.assert?.(t, a, h);
            }
        });
        return 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.enqueue(ref.relay, t =>
            _sync(t, ref.target, this.ref({ message: resolve }))));
    }

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

    enqueue(relay: Actor, a: LocalAction): void {
        this.queues.get(relay)?.push(a) ?? this.queues.set(relay, [a]);
    }
}

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

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

    function walk(p: Pattern, v: Assertion): boolean {
        return fold(p, {
            boolean(_b) { return is(p, v); },
            single(_f) { return is(p, v); },
            double(_f) { return is(p, v); },
            integer(_i) { return is(p, v); },
            string(_s) { return is(p, v); },
            bytes(_b) { return is(p, v); },
            symbol(_s) { return is(p, v); },
            record(r) {
                v = unannotate(v);
                return Record.isRecord<Ref>(v)
                    && r.length === v.length
                    && walk(r.label, v.label)
                    && r.every((pp, i) => walk(pp, (v as Record<Ref>)[i]));
            },
            array(a) {
                v = unannotate(v);
                return Array.isArray(v)
                    && a.length === v.length
                    && a.every((pp, i) =>
                        walk(pp, (v as Array<Value<Ref>>)[i]));
            },
            set(_s) { return false; }, // TODO: set patterns
            dictionary(d) {
                v = unannotate(v);
                if (!Dictionary.isDictionary<Ref>(v)) return false;
                if (d.size !== v.size) return false;
                for (const pe of d.entries()) {
                    const [pk, pv] = pe;
                    if (!v.has(pk)) return false;
                    if (!walk(pv, v.get(pk)!)) return false;
                }
                return true;
            },
            annotated(a) { return walk(a.item, v); },
            pointer(t) {
                if ('type' in t) {
                    switch (t.type) {
                        case 'bind':
                            bindings[t.name] = v;
                            return walk(t.pattern, v);
                        case 'discard':
                            return true;
                        case 'not':
                            return !walk(t.pattern, v);
                        case 'and':
                            for (const p of t.patterns) {
                                if (!walk(p, v)) return false;
                            }
                            return true;
                        case 'or':
                            for (const p of t.patterns) {
                                if (walk(p, v)) return true;
                            }
                            return false;
                    }
                } else {
                    if (!isRef(v)) return false;
                    return (t.relay === v.relay) && (t.target === v.target);
                }
            },
        });
    }

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

export function instantiate(t: Template, b: Bindings): Assertion {
    return mapPointers(t, p => ('type' in p) ? b[p.name] : p);
}

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 runRewrites(a: Attenuation, v: Assertion): Assertion | null {
    for (const r of a) {
        const w = rewrite(r, v);
        if (w === null) return null;
        v = w;
    }
    return v;
}

export function rfilter(... patterns: Pattern[]): Rewrite {
    return { pattern: pbind('a', por(... patterns)), template: tref('a') };
}

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