Saturday, June 6, 2026

from dataclasses import dataclass
from enum import Enum
from typing import Optional

class OpType(Enum):
    INSERT = "insert"
    DELETE = "delete"
    RETAIN = "retain"  # skip N characters unchanged

@dataclass
class Operation:
    """
    Represents a single atomic edit operation.
    Uses the 'retain/insert/delete' triple common in OT libraries (like OT.js / ShareDB).
    """
    type: OpType
    pos: int          # position in the ORIGINAL document (before this op)
    char: Optional[str] = None  # only for INSERT
    length: int = 1   # for DELETE or RETAIN


def transform(op_a: Operation, op_b: Operation, priority: str = "left") -> Operation:
    """
    Transform op_a against op_b.
    priority="left" means op_a wins on tie (same position insert).
    
    This implements the INSERT/DELETE x INSERT/DELETE matrix.
    """
    if op_a.type == OpType.INSERT and op_b.type == OpType.INSERT:
        # Both insert at same position: priority breaks tie
        if op_b.pos < op_a.pos or (op_b.pos == op_a.pos and priority == "right"):
            # op_b inserts before (or at same pos with lower priority)
            # → op_a's position shifts right by 1
            return Operation(OpType.INSERT, op_a.pos + 1, op_a.char)
        return op_a  # op_b is at or after op_a, no shift needed

    elif op_a.type == OpType.INSERT and op_b.type == OpType.DELETE:
        if op_b.pos < op_a.pos:
            # op_b deleted something before our insert position
            # → our position shifts left by 1
            return Operation(OpType.INSERT, op_a.pos - 1, op_a.char)
        return op_a  # deletion was at or after our position, no shift

    elif op_a.type == OpType.DELETE and op_b.type == OpType.INSERT:
        if op_b.pos <= op_a.pos:
            # op_b inserted before our deletion target
            # → our target shifts right by 1
            return Operation(OpType.DELETE, op_a.pos + 1)
        return op_a

    elif op_a.type == OpType.DELETE and op_b.type == OpType.DELETE:
        if op_b.pos < op_a.pos:
            # op_b deleted something before our target → shift left
            return Operation(OpType.DELETE, op_a.pos - 1)
        elif op_b.pos == op_a.pos:
            # Both deleting the same character! op_a becomes a no-op.
            # We return a RETAIN(0) to signal "already done"
            return Operation(OpType.RETAIN, op_a.pos, length=0)
        return op_a


class OTDocument:
    """
    Simple OT-based collaborative document (server-side coordinator).
    """
    def __init__(self):
        self.text = ""
        self.history: list[Operation] = []  # all committed ops in order

    def apply(self, op: Operation) -> str:
        """Apply an operation to the document text."""
        if op.type == OpType.INSERT:
            self.text = self.text[:op.pos] + op.char + self.text[op.pos:]
        elif op.type == OpType.DELETE:
            if op.length > 0:  # skip no-op deletes
                self.text = self.text[:op.pos] + self.text[op.pos + 1:]
        return self.text

    def receive(self, op: Operation, client_version: int) -> Operation:
        """
        Server receives an op from a client.
        client_version = how many ops the client has seen (its 'parent' revision).
        
        We must transform op against all ops since client_version.
        """
        # Transform op against every concurrent op the client hasn't seen yet
        concurrent_ops = self.history[client_version:]
        transformed_op = op
        for server_op in concurrent_ops:
            transformed_op = transform(transformed_op, server_op)
        
        self.apply(transformed_op)
        self.history.append(transformed_op)
        return transformed_op  # broadcast this to all other clients


# ─── Demo: The classic conflict scenario ───────────────────────────────────────
doc = OTDocument()
doc.text = "abc"
doc.history = []

# User A: insert "X" at position 0
op_a = Operation(OpType.INSERT, pos=0, char="X")

# User B: delete 'c' at position 2
op_b = Operation(OpType.DELETE, pos=2)

# A's op arrives first at version 0 → applied directly
doc.receive(op_a, client_version=0)
print(f"After A: '{doc.text}'")  # "Xabc"

# B's op arrives (still thinks it's at version 0, before A)
# server transforms B against A's insert
doc.receive(op_b, client_version=0)
print(f"After B: '{doc.text}'")  # "Xab" ← 'c' correctly deleted ✓

import uuid
from dataclasses import dataclass, field
from typing import Optional

@dataclass(order=True)
class Identifier:
    """
    Every character in a CRDT document gets a globally unique, 
    totally-ordered identifier. No two identifiers are equal.
    
    This is the key insight: positions become stable IDs, not integers.
    When A inserts between IDs 10 and 20, it picks ID 15.
    No matter what B does concurrently, character 15 stays where it is.
    """
    pos: float       # fractional position for ordering (Logoot-style)
    site_id: str     # which client created this (for tiebreaking)
    counter: int     # lamport clock to ensure uniqueness
    
    def __eq__(self, other):
        return self.pos == other.pos and self.site_id == other.site_id


@dataclass
class CRDTChar:
    """A single character in the CRDT sequence."""
    id: Identifier
    char: str
    deleted: bool = False  # tombstone — deleted chars stay in the structure


class RGADocument:
    """
    Replicated Growable Array (RGA) — a popular CRDT for text.
    
    Key properties:
    - Each char has a unique ID (site_id + lamport clock)
    - Causality: an insert references its left neighbor's ID
    - Delete = tombstone (never physically remove, just mark)
    - Merge: union of all char sets, order by ID
    """
    def __init__(self, site_id: str):
        self.site_id = site_id
        self.clock = 0
        # chars sorted by their stable IDs
        self._chars: list[CRDTChar] = []

    def _new_id(self, left_pos: float, right_pos: float) -> Identifier:
        """
        Generate a new identifier that falls between left and right positions.
        Uses fractional indexing — the classic "pick a number between 10 and 20".
        """
        self.clock += 1
        mid_pos = (left_pos + right_pos) / 2
        return Identifier(pos=mid_pos, site_id=self.site_id, counter=self.clock)

    def local_insert(self, char: str, after_index: int) -> CRDTChar:
        """
        Insert a character after the visible character at after_index.
        Returns the new CRDTChar (to be broadcast to peers).
        """
        # Find the fractional position boundaries
        left_pos = self._chars[after_index].id.pos if after_index >= 0 else 0.0
        right_pos = (
            self._chars[after_index + 1].id.pos 
            if after_index + 1 < len(self._chars) 
            else 1.0
        )
        new_id = self._new_id(left_pos, right_pos)
        new_char = CRDTChar(id=new_id, char=char)
        
        # Insert in sorted position
        insert_at = after_index + 1
        self._chars.insert(insert_at, new_char)
        return new_char

    def local_delete(self, index: int) -> CRDTChar:
        """
        Delete the visible character at index.
        Tombstone only — the char stays in _chars forever.
        """
        # Find the nth visible (non-deleted) char
        count = 0
        for c in self._chars:
            if not c.deleted:
                if count == index:
                    c.deleted = True
                    return c  # broadcast this tombstone
                count += 1
        raise IndexError("No visible character at that index")

    def apply_remote(self, op: CRDTChar):
        """
        Merge a remote operation (insert or delete) into our local state.
        This is the magic: it's commutative, associative, idempotent.
        """
        if op.deleted:
            # Remote delete: find and tombstone our local copy
            for c in self._chars:
                if c.id == op.id:
                    c.deleted = True
                    return
        else:
            # Remote insert: find the right sorted position by ID
            for i, c in enumerate(self._chars):
                if c.id > op.id:
                    self._chars.insert(i, op)
                    return
            self._chars.append(op)

    @property
    def text(self) -> str:
        """Return the visible (non-tombstoned) text."""
        return "".join(c.char for c in self._chars if not c.deleted)


# ─── Demo: Same conflict, CRDT way ─────────────────────────────────────────────
doc_a = RGADocument(site_id="A")
doc_b = RGADocument(site_id="B")

# Both start with "abc" — in a real system, initial state is synced
# (simplified: skip initial state sync for clarity)

# The point: concurrent operations just get merged, no transform needed.
# doc_a.local_insert("X", 0) → broadcasts CRDTChar with ID pos~0.5
# doc_b.local_delete(2)       → broadcasts tombstone for 'c'
# Both sides call apply_remote() with the other's op
# Result is always identical regardless of order ✓

# 问题：大量删除后，_chars 越来越大
# "Hello World" typed then deleted → 11 tombstones remain in memory
# 
# 解法：定期 Compaction（需要全局协调）
# 只有当所有客户端都 >= 某个版本时，才能安全清除 tombstone
# Google Docs 通过定期 snapshot + log replay 解决这个问题

CRDT wins here: offline edits accumulate as local ops.
On reconnect, just apply_remote() all the remote ops.
No conflict — the math guarantees convergence.

OT requires the server version number, so offline OT is tricky.
Google Docs handles this by queuing ops locally and replaying on reconnect.

# 用户意图："删除第二个词"
# 并发操作可能在删除之前插入了一堆内容
# OT 变换后删除了"正确位置"的字符，但可能不是用户想删的那个词
#
# 这是 OT 最深的哲学问题，没有完美解法
# 实践中：用光标 heuristics + 用户 undo 来缓解