suaveplan/packages/foundation/flow/crdt/tests/integration.test.ts

import { describe, expect, it } from "@suaveplan/testing/runner";
import { CrdtSyncManager } from "../src/core/sync-manager/sync-manager.js";
import { GCounter } from "../src/crdts/g-counter/g-counter.js";
import { LWWRegister } from "../src/crdts/lww-register/lww-register.js";
import { MVRegister } from "../src/crdts/mv-register/mv-register.js";
import { ORSet } from "../src/crdts/or-set/or-set.js";
import { PNCounter } from "../src/crdts/pn-counter/pn-counter.js";
import { RGA } from "../src/crdts/rga/rga.js";
import { TwoPSet } from "../src/crdts/two-p-set/two-p-set.js";

describe("CRDT Integration", () => {
    it("GCounter: all replicas converge after three-way sync", {
        timeout: 5000,
    }, () => {
        const a = new GCounter("shared");
        const b = new GCounter("shared");
        const c = new GCounter("shared");

        a.increment("a", 1);
        b.increment("b", 2);
        c.increment("c", 3);

        a.merge(b.state());
        a.merge(c.state());
        b.merge(a.state());
        c.merge(a.state());

        expect(a.total()).toBe(6);
        expect(b.total()).toBe(6);
        expect(c.total()).toBe(6);
    });

    it("PNCounter: bidirectional increments and decrements", {
        timeout: 5000,
    }, () => {
        const a = new PNCounter("shared");
        const b = new PNCounter("shared");

        a.increment("a", 10);
        b.decrement("b", 4);

        a.merge(b.state());
        b.merge(a.state());

        expect(a.total()).toBe(6);
        expect(b.total()).toBe(6);
    });

    it("ORSet: concurrent add/remove converges", { timeout: 5000 }, () => {
        const a = new ORSet<string>("s1");
        const b = new ORSet<string>("s1");

        a.add("x", "a");
        b.merge(a.state());
        b.add("x", "b"); // new unique tag for same element
        a.remove("x"); // tombstones a's tag

        a.merge(b.state());

        // b's tag is still alive — element should be present
        expect(a.has("x")).toBe(true);
    });

    it("LWWRegister: last write wins", { timeout: 5000 }, () => {
        const a = new LWWRegister<string>("nodeA", "r1");
        const b = new LWWRegister<string>("nodeB", "r1");

        a.set("old", 100);
        b.set("new", 200);

        a.merge(b.state());
        expect(a.value()).toBe("new");
    });

    it("MVRegister: concurrent writes preserved", { timeout: 5000 }, () => {
        const a = new MVRegister<string>("nodeA", "r1");
        const b = new MVRegister<string>("nodeB", "r1");

        a.set("val-a");
        b.set("val-b");

        a.merge(b.state());

        const vals = a.get();
        expect(vals).toContain("val-a");
        expect(vals).toContain("val-b");
    });

    it("TwoPSet: removal prevents re-add", { timeout: 5000 }, () => {
        const s = new TwoPSet<string>();
        s.add("item");
        s.remove("item");
        expect(() => s.add("item")).toThrow();
    });

    it("RGA: three-replica document convergence", { timeout: 5000 }, () => {
        const a = new RGA<string>("a", "doc");
        const b = new RGA<string>("b", "doc");
        const c = new RGA<string>("c", "doc");

        const r1 = a.insert("A");
        b.merge(a.state());
        c.merge(a.state());

        a.insert("X", r1);
        b.insert("Y", r1);
        c.insert("Z", r1);

        a.merge(b.state());
        a.merge(c.state());
        b.merge(a.state());
        c.merge(a.state());

        expect(a.value()).toEqual(b.value());
        expect(b.value()).toEqual(c.value());
        expect(a.value()).toHaveLength(4);
    });

    it("SyncManager: multi-CRDT sync across replicas", {
        timeout: 5000,
    }, () => {
        const mgrA = new CrdtSyncManager({ nodeId: "a" });
        const mgrB = new CrdtSyncManager({ nodeId: "b" });

        const counterA = new GCounter("counter");
        const counterB = new GCounter("counter");
        const setA = new ORSet<string>("set");
        const setB = new ORSet<string>("set");

        mgrA.register(counterA);
        mgrA.register(setA);
        mgrB.register(counterB);
        mgrB.register(setB);

        counterA.increment("a", 5);
        setA.add("hello", "a");
        mgrA.tick();

        counterB.increment("b", 3);
        mgrB.tick();

        mgrA.syncWith(mgrB);

        expect(counterA.total()).toBe(8);
        expect(counterB.total()).toBe(8);
        expect(setA.has("hello")).toBe(true);
        expect(setB.has("hello")).toBe(true);
    });

    it("JSON round-trip for all CRDT types", { timeout: 5000 }, () => {
        const counter = new GCounter("n", "c1");
        counter.increment("n", 42);
        const c2 = new GCounter("n", "c1");
        c2.fromJSON(counter.toJSON());
        expect(c2.total()).toBe(42);

        const set = new ORSet<string>("s1");
        set.add("foo", "n");
        const s2 = new ORSet<string>("s1");
        s2.fromJSON(set.toJSON());
        expect(s2.has("foo")).toBe(true);

        const rga = new RGA<number>("n", "r1");
        rga.insert(99);
        const r2 = new RGA<number>("n", "r1");
        r2.fromJSON(rga.toJSON());
        expect(r2.value()).toContain(99);
    });
});