Add vector clock, oplog hashing & chain sync

Introduce vector clock support and oplog hashing to improve sync integrity and gap recovery. Added HLCTimestamp-based VectorClock class and computeOplogHash utility; extended core exports. Updated storage interface and SqlitePeerStore to persist hash/previous_hash, compute entry hashes on write, expose vector clock, per-node oplog queries, last-entry hash and chain-range retrieval, and map DB rows to OplogEntry. Extended protocol and protobuf (sync.proto + generated) with VectorClock messages, hash/previous_hash on ProtoOplogEntry, GetChainRange/GetVectorClock messages and message types. Updated ProtocolMapper and domain OplogEntry to carry hash fields. Enhanced TcpSyncClient with getVectorClock, pushChanges and getChainRange helpers. Revamped SyncOrchestrator to exchange vector clocks, compute nodes to pull/push, process inbound batches with chain verification and gap recovery, and apply changes atomically. These changes enable per-node incremental sync, integrity checks and missing-range recovery.

Author: mrdevrobot

packages/core/src/crypto/hash.ts

@@ -0,0 +1,84 @@
+import { createHash } from 'crypto';
+import { OplogEntry, HLCTimestamp } from '@entgldb/protocol';
+import { HLClock } from '../hlc/clock';
+
+export function computeOplogHash(entry: OplogEntry): string {
+    const sha256 = createHash('sha256');
+    const sb: string[] = [];
+
+    // Order: Collection|Key|Operation|Payload|Timestamp|PreviousHash
+    // Operation is enum (0=Put, 1=Delete). Node uses exact same proto enum.
+
+    sb.push(entry.collection);
+    sb.push('|');
+    sb.push(entry.key);
+    sb.push('|');
+    // Operation is number in protobuf-ts interface usually? Or string?
+    // In generated code, enum is usually a number.
+    // Check .NET: it appends 'Operation' which is an Enum. 
+    // In C# ToString() on Enum returns the name "Put" unless [Flags] or cast.
+    // WAIT. .NET code: 'sb.Append(Operation);'
+    // Step 14: public OperationType Operation { get; }
+    // public enum OperationType { Put, Delete }
+    // In C#, sb.Append(enum) appends the string name! "Put", "Delete".
+    // Protobuf usually sends integers 0, 1.
+    // I need to be careful here. 
+    // If .NET computes hash using "Put", I must use "Put".
+    // If I use the number 0, hashes will differ.
+    // I should check OplogEntry.cs again.
+    // Yes, 'sb.Append(Operation)' where Operation is OperationType enum.
+    // So it uses names.
+
+    // In Node.js, the generated proto interface likely uses numbers or string literals depending on option.
+    // But OplogEntry is the database object, not necessarily the proto message yet?
+    // In Node, we often use the Proto interface as the domain object or similar.
+    // Let's assume input 'entry' is the ProtoOplogEntry-like structure.
+    // If 'operation' is "Put" or "Delete" string, good. 
+    // If it's number 0, 1, I need to map it.
+
+    // Let's check sync.proto again.
+    // message ProtoOplogEntry { string operation = 3; } -> Wait!
+    // In sync.proto (Step 104/106), 'string operation = 3; // "Put" or "Delete"'
+    // Ah, in proto it is defined as STRING. 
+    // In .NET OplogEntry.cs, it's an Enum, but mapped to string in proto?
+    // Step 13 (Net proto): string operation = 3;
+    // Step 14 (Net logic): public OperationType Operation { get; } -> SB.Append(Operation).
+    // So the hash source is the Enum name "Put"/"Delete".
+    // The proto carries a string "Put"/"Delete".
+    // So if I use the string from the proto, it should match.
+
+    // BUT, wait.
+    // In Node.js, is OplogEntry type using string or enum?
+    // Step 71: export type { OplogEntry } from '@entgldb/protocol';
+    // If I just updated sync.proto to have `string operation = 3`, then it's string.
+    // Previously in Node proto? Step 104: `string operation = 3`.
+    // So it's string.
+
+    // OplogEntry domain object has 'data' as Uint8Array, 'timestamp' as object.
+
+    // Operation
+    sb.push(entry.operation);
+    sb.push('|');
+
+    // Payload
+    if (entry.data && entry.data.length > 0) {
+        // Decode bytes to string for hashing to match .NET GetRawText()
+        // Assuming data is UTF-8 JSON
+        sb.push(new TextDecoder().decode(entry.data));
+    }
+    sb.push('|');
+
+    // Timestamp
+    if (entry.timestamp) {
+        sb.push(HLClock.toString(entry.timestamp));
+    } else {
+        // Should not happen for valid entry
+        sb.push('0-0-');
+    }
+    sb.push('|');
+
+    sb.push(entry.previousHash || '');
+
+    sha256.update(sb.join(''));
+    return sha256.digest('hex').toLowerCase();
+}

packages/core/src/hlc/vector-clock.ts

@@ -0,0 +1,133 @@
+import { HLCTimestamp } from '@entgldb/protocol';
+import { HLClock } from './clock';
+
+export enum CausalityRelation {
+    Equal,
+    StrictlyAhead,
+    StrictlyBehind,
+    Concurrent
+}
+
+export class VectorClock {
+    private clock: Map<string, HLCTimestamp>;
+
+    constructor(initial?: Map<string, HLCTimestamp>) {
+        this.clock = initial ? new Map(initial) : new Map();
+    }
+
+    get nodeIds(): string[] {
+        return Array.from(this.clock.keys());
+    }
+
+    getTimestamp(nodeId: string): HLCTimestamp | undefined {
+        return this.clock.get(nodeId);
+    }
+
+    setTimestamp(nodeId: string, timestamp: HLCTimestamp): void {
+        this.clock.set(nodeId, timestamp);
+    }
+
+    merge(other: VectorClock): void {
+        for (const nodeId of other.nodeIds) {
+            const otherTs = other.getTimestamp(nodeId)!;
+            const currentTs = this.getTimestamp(nodeId);
+
+            if (!currentTs || HLClock.compare(otherTs, currentTs) > 0) {
+                this.setTimestamp(nodeId, otherTs);
+            }
+        }
+    }
+
+    compareTo(other: VectorClock): CausalityRelation {
+        let thisAhead = false;
+        let otherAhead = false;
+
+        const allNodes = new Set([...this.nodeIds, ...other.nodeIds]);
+
+        for (const nodeId of allNodes) {
+            const thisTs = this.getTimestamp(nodeId);
+            const otherTs = other.getTimestamp(nodeId);
+
+            // If a node is missing, treat as "zero" timestamp (which is always older than any real timestamp)
+            // But strict implementation might require explicit handling. 
+            // Assuming missing means "start of time".
+            
+            let cmp = 0;
+            if (thisTs && otherTs) {
+                cmp = HLClock.compare(thisTs, otherTs);
+            } else if (thisTs && !otherTs) {
+                cmp = 1; 
+            } else if (!thisTs && otherTs) {
+                cmp = -1;
+            }
+
+            if (cmp > 0) thisAhead = true;
+            if (cmp < 0) otherAhead = true;
+
+            if (thisAhead && otherAhead) return CausalityRelation.Concurrent;
+        }
+
+        if (thisAhead && !otherAhead) return CausalityRelation.StrictlyAhead;
+        if (otherAhead && !thisAhead) return CausalityRelation.StrictlyBehind;
+        return CausalityRelation.Equal;
+    }
+
+    getNodesWithUpdates(other: VectorClock): string[] {
+        const result: string[] = [];
+        const allNodes = new Set(this.clock.keys());
+        
+        // Add nodes that other has but we don't (implicitly other is ahead)
+        for (const nodeId of other.nodeIds) {
+            if (!this.clock.has(nodeId)) {
+                // Actually if other has a node we don't, other is ahead for that node
+                // But usually we track "known nodes". 
+                // Wait, logic: "Nodes where OTHER is ahead of THIS"
+                // So if I don't have it, and other has it, other is ahead.
+                allNodes.add(nodeId);
+            }
+        }
+
+        for (const nodeId of allNodes) {
+            const thisTs = this.getTimestamp(nodeId);
+            const otherTs = other.getTimestamp(nodeId);
+
+            // If other has it and (we don't OR other > us)
+            if (otherTs) {
+                if (!thisTs || HLClock.compare(otherTs, thisTs) > 0) {
+                    result.push(nodeId);
+                }
+            }
+        }
+        return result;
+    }
+
+    getNodesToPush(other: VectorClock): string[] {
+        const result: string[] = [];
+        const allNodes = new Set([...this.nodeIds, ...other.nodeIds]);
+
+        for (const nodeId of allNodes) {
+            const thisTs = this.getTimestamp(nodeId);
+            const otherTs = other.getTimestamp(nodeId);
+
+            // If we have it and (other doesn't OR we > other)
+            if (thisTs) {
+                if (!otherTs || HLClock.compare(thisTs, otherTs) > 0) {
+                    result.push(nodeId);
+                }
+            }
+        }
+        return result;
+    }
+
+    clone(): VectorClock {
+        // deepish clone of map (timestamps are immutable-ish usually)
+        return new VectorClock(this.clock);
+    }
+
+    toString(): string {
+        if (this.clock.size === 0) return '{}';
+        const entries = Array.from(this.clock.entries())
+            .map(([k, v]) => `${k}:${HLClock.toString(v)}`);
+        return `{${entries.join(', ')}}`;
+    }
+}

packages/core/src/index.ts

@@ -10,6 +10,8 @@ export { IPeerStore } from './storage/interface';
 
 // Sync and conflict resolution
 export * from './sync';
+export * from './hlc/vector-clock';
+export * from './crypto/hash';
 
 // Re-export protocol types for convenience
 // Re-export protocol types for convenience

packages/core/src/storage/interface.ts

@@ -31,11 +31,31 @@ export interface IPeerStore {
      */
     deleteDocument(collection: string, key: string, timestamp: HLCTimestamp): Promise<void>;
 
+    /**
+     * Get the current Vector Clock from the store
+     */
+    getVectorClock(): Promise<import('../hlc/vector-clock').VectorClock>;
+
     /**
      * Get oplog entries after a given timestamp
      */
     getOplogAfter(timestamp: HLCTimestamp, limit?: number): Promise<OplogEntry[]>;
 
+    /**
+     * Get oplog entries for a specific node after a given timestamp
+     */
+    getOplogForNodeAfter(nodeId: string, timestamp: HLCTimestamp): Promise<OplogEntry[]>;
+
+    /**
+     * Get the hash of the last entry for a specific node
+     */
+    getLastEntryHash(nodeId: string): Promise<string | null>;
+
+    /**
+     * Get a range of oplog entries between two hashes (inclusive end)
+     */
+    getChainRange(startHash: string, endHash: string): Promise<OplogEntry[]>;
+
     /**
      * Apply a batch of documents and oplog entries (for sync)
      */