novy-syndicate/actor.ts

import { IdentitySet, Value } from 'preserves';

export type Assertion = Value<Ref>;

export type Handle = number;

export type ExitReason = null | { ok: true } | { ok: false, err: Error };

export const assert = Symbol('assert');
export const retract = Symbol('retract');
export const message = Symbol('message');
export const sync = Symbol('sync');

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: Entity): void;
}

export class Ref implements Entity {
    readonly relay: Actor;
    readonly target: Entity;

    constructor(relay: Actor, target: Entity) {
        this.relay = relay;
        this.target = target;
    }

    [assert](turn: Turn, assertion: Assertion, handle: Handle): void {
        this.target[assert]?.(turn, assertion, handle);
    }

    [retract](turn: Turn, handle: Handle): void {
        this.target[retract]?.(turn, handle);
    }

    [message](turn: Turn, body: Assertion): void {
        this.target[message]?.(turn, body);
    }

    [sync](turn: Turn, peer: Ref) {
        this.target[sync] ? this.target[sync]!(turn, peer) : turn.message(peer, true);
    }
}

export type OutboundMap = Map<Handle, Ref>;

export class Actor {
    readonly outbound: OutboundMap;
    exitReason: ExitReason = null;

    constructor(initialAssertions: OutboundMap = new Map()) {
        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;

type LocalAction = (t: Turn) => void;

export class Turn {
    readonly actor: Actor;
    readonly queues: Map<Actor, LocalAction[]> = 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))));
    }

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

    ref(e: Entity): Ref {
        return (e instanceof Ref) ? e : new Ref(this.actor, e);
    }

    spawn(bootProc: LocalAction, initialAssertions = new IdentitySet<Handle>()): void {
        this.enqueue(this.actor, () => {
            const newOutbound: OutboundMap = new Map();
            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 => {
            this.actor.outbound.set(h, ref);
            ref[assert]?.(t, assertion, h);
        });
        return h;
    }

    retract(h: Handle): void {
        this._retract(this.actor.outbound.get(h)!, 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[retract]?.(t, handle);
        });
    }

    sync(ref: Ref): Promise<Turn> {
        return new Promise(resolve =>
            this.enqueue(ref.relay, t => ref[sync]?.(t, this.ref({ [message]: resolve }))));
    }

    message(ref: Ref, assertion: Assertion): void {
        this.enqueue(ref.relay, t => ref[message]?.(t, assertion));
    }

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