প্রজেক্ট: রিয়েল-টাইম কোলাবোরেটিভ এডিটর

package crdt

import (
	"fmt"
)

// RGA (Replicated Growable Array) is a sequence CRDT for collaborative text.
//
// Every character carries an immutable id. Inserts reference the id of the
// character they follow, never an index, so concurrent edits elsewhere in the
// document can never make an operation land in the wrong place.

type OpID struct {
	Site    string `json:"site"`
	Counter int64  `json:"counter"`
}

func (id OpID) Key() string {
	return fmt.Sprintf("%s:%d", id.Site, id.Counter)
}

// Greater is the deterministic total order used to break ties between
// concurrent inserts at the same anchor. Every replica applies the identical
// rule, so no coordination is needed.
func (id OpID) Greater(other OpID) bool {
	if id.Counter != other.Counter {
		return id.Counter > other.Counter
	}
	return id.Site > other.Site
}

type OpType string

const (
	OpInsert OpType = "insert"
	OpDelete OpType = "delete"
)

type Op struct {
	Type   OpType `json:"type"`
	ID     OpID   `json:"id"`
	After  *OpID  `json:"after,omitempty"`  // insert only
	Value  string `json:"value,omitempty"`  // insert only
	Target *OpID  `json:"target,omitempty"` // delete only
}

type node struct {
	id      OpID
	value   string
	deleted bool
	after   *OpID
	next    *node
}

type RGA struct {
	site    string
	head    *node
	index   map[string]*node
	counter int64
	// seen is the version vector: the highest counter observed per site.
	seen map[string]int64
	// pending holds ops that arrived before their anchor, keyed by anchor.
	pending map[string][]Op
}

func New(site string) *RGA {
	head := &node{id: OpID{Site: "", Counter: 0}, deleted: true}
	r := &RGA{
		site:    site,
		head:    head,
		index:   map[string]*node{head.id.Key(): head},
		seen:    map[string]int64{},
		pending: map[string][]Op{},
	}
	return r
}

// --- Local editing API (produces ops to broadcast) ---

// LocalInsert inserts text at the given visible offset and returns the ops to
// broadcast to other replicas.
func (r *RGA) LocalInsert(offset int, text string) []Op {
	anchor := r.nodeAtVisibleOffset(offset)
	ops := make([]Op, 0, len(text))
	for _, ch := range text {
		r.counter++
		id := OpID{Site: r.site, Counter: r.counter}
		var after *OpID
		if anchor != r.head {
			a := anchor.id
			after = &a
		}
		op := Op{Type: OpInsert, ID: id, After: after, Value: string(ch)}
		r.applyInsert(op)
		ops = append(ops, op)
		anchor = r.index[id.Key()]
	}
	return ops
}

// LocalDelete removes length visible characters starting at offset.
func (r *RGA) LocalDelete(offset, length int) []Op {
	var ops []Op
	start := r.nodeAtVisibleOffset(offset)
	cur := start.next
	removed := 0
	for cur != nil && removed < length {
		if !cur.deleted {
			r.counter++
			target := cur.id
			op := Op{
				Type:   OpDelete,
				ID:     OpID{Site: r.site, Counter: r.counter},
				Target: &target,
			}
			r.applyDelete(op)
			ops = append(ops, op)
			removed++
		}
		cur = cur.next
	}
	return ops
}

// --- Remote application ---

// ApplyRemote applies an op from another replica. It is idempotent (a replayed
// op is a no-op) and tolerant of out-of-order arrival (an op whose anchor is
// missing is buffered until the anchor shows up).
func (r *RGA) ApplyRemote(op Op) error {
	if r.alreadySeen(op) {
		return nil
	}

	switch op.Type {
	case OpInsert:
		anchorKey := r.head.id.Key()
		if op.After != nil {
			anchorKey = op.After.Key()
		}
		if _, ok := r.index[anchorKey]; !ok {
			r.buffer(anchorKey, op)
			return nil
		}
		r.applyInsert(op)
		r.drainPending(op.ID.Key())
	case OpDelete:
		if op.Target == nil {
			return fmt.Errorf("delete op %s has no target", op.ID.Key())
		}
		if _, ok := r.index[op.Target.Key()]; !ok {
			r.buffer(op.Target.Key(), op)
			return nil
		}
		r.applyDelete(op)
	default:
		return fmt.Errorf("unknown op type %q", op.Type)
	}
	return nil
}

func (r *RGA) applyInsert(op Op) {
	anchorKey := r.head.id.Key()
	if op.After != nil {
		anchorKey = op.After.Key()
	}
	prev, ok := r.index[anchorKey]
	if !ok {
		panic("missing anchor " + anchorKey)
	}

	// Skip over concurrent inserts at the same anchor that sort higher than
	// ours. Everyone applies the same comparison, so everyone converges.
	for prev.next != nil && sameAnchor(prev.next.after, op.After) && prev.next.id.Greater(op.ID) {
		prev = prev.next
	}

	n := &node{id: op.ID, value: op.Value, after: op.After, next: prev.next}
	prev.next = n
	r.index[op.ID.Key()] = n

	if op.ID.Site == r.site && op.ID.Counter > r.counter {
		r.counter = op.ID.Counter
	}
	r.markSeen(op.ID)
}

func (r *RGA) applyDelete(op Op) {
	n := r.index[op.Target.Key()]
	// Tombstone, never unlink: other replicas may still anchor to this node.
	n.deleted = true
	r.markSeen(op.ID)
}

func sameAnchor(a, b *OpID) bool {
	if a == nil || b == nil {
		return a == nil && b == nil
	}
	return a.Site == b.Site && a.Counter == b.Counter
}

// --- Causality bookkeeping ---

func (r *RGA) alreadySeen(op Op) bool {
	return r.seen[op.ID.Site] >= op.ID.Counter
}

func (r *RGA) markSeen(id OpID) {
	if id.Counter > r.seen[id.Site] {
		r.seen[id.Site] = id.Counter
	}
}

func (r *RGA) buffer(missingKey string, op Op) {
	r.pending[missingKey] = append(r.pending[missingKey], op)
}

func (r *RGA) drainPending(nowAvailableKey string) {
	waiting, ok := r.pending[nowAvailableKey]
	if !ok {
		return
	}
	delete(r.pending, nowAvailableKey)
	for _, op := range waiting {
		_ = r.ApplyRemote(op)
	}
}

// VersionVector reports what this replica has already seen, per site.
func (r *RGA) VersionVector() map[string]int64 {
	out := make(map[string]int64, len(r.seen))
	for k, v := range r.seen {
		out[k] = v
	}
	return out
}

// --- Reading ---

func (r *RGA) String() string {
	var buf []byte
	for n := r.head.next; n != nil; n = n.next {
		if !n.deleted {
			buf = append(buf, n.value...)
		}
	}
	return string(buf)
}

func (r *RGA) nodeAtVisibleOffset(offset int) *node {
	cur := r.head
	seen := 0
	for cur.next != nil && seen < offset {
		cur = cur.next
		if !cur.deleted {
			seen++
		}
	}
	return cur
}

// --- Snapshotting ---

type SnapshotNode struct {
	ID      OpID   `json:"id"`
	Value   string `json:"value"`
	Deleted bool   `json:"deleted"`
	After   *OpID  `json:"after,omitempty"`
}

type Snapshot struct {
	Site    string         `json:"site"`
	Counter int64          `json:"counter"`
	Nodes   []SnapshotNode `json:"nodes"`
}

// Snapshot keeps every node, including tombstones, because anchors must
// survive. Compaction (dropping tombstones nobody can reference any more) is
// only safe once every site has acknowledged past them.
func (r *RGA) Snapshot() Snapshot {
	snap := Snapshot{Site: r.site, Counter: r.counter}
	for n := r.head.next; n != nil; n = n.next {
		snap.Nodes = append(snap.Nodes, SnapshotNode{
			ID: n.id, Value: n.value, Deleted: n.deleted, After: n.after,
		})
	}
	return snap
}

func FromSnapshot(site string, snap Snapshot) *RGA {
	r := New(site)
	prev := r.head
	for _, sn := range snap.Nodes {
		n := &node{id: sn.ID, value: sn.Value, deleted: sn.Deleted, after: sn.After}
		prev.next = n
		r.index[sn.ID.Key()] = n
		if sn.ID.Counter > r.seen[sn.ID.Site] {
			r.seen[sn.ID.Site] = sn.ID.Counter
		}
		prev = n
	}
	r.counter = snap.Counter
	return r
}

package room

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"example.com/crdt"
)

// --- Wire protocol ---

type ClientMessage struct {
	T     string           `json:"t"` // hello | push | awareness | ping
	DocID string           `json:"docId,omitempty"`
	Site  string           `json:"siteId,omitempty"`
	VV    map[string]int64 `json:"vv,omitempty"`
	Ops   []crdt.Op        `json:"ops,omitempty"`
	State *AwarenessState  `json:"state,omitempty"`
}

type ServerMessage struct {
	T           string            `json:"t"` // sync | ops | awareness | error | pong
	SnapshotURL string            `json:"snapshotUrl,omitempty"`
	Ops         []crdt.Op         `json:"ops,omitempty"`
	VV          map[string]int64  `json:"vv,omitempty"`
	Peers       []AwarenessPeer   `json:"peers,omitempty"`
	Code        string            `json:"code,omitempty"`
	Message     string            `json:"message,omitempty"`
}

type Cursor struct {
	Anchor *crdt.OpID `json:"anchor"`
	Head   *crdt.OpID `json:"head"`
}

type AwarenessState struct {
	DisplayName string  `json:"displayName"`
	Cursor      *Cursor `json:"cursor"`
	Colour      string  `json:"colour"`
}

type AwarenessPeer struct {
	SiteID string `json:"siteId"`
	AwarenessState
}

type Connection interface {
	SiteID() string
	UserID() string
	Send(ServerMessage)
	Close(reason string)
}

// OpLog is the durable source of truth for a document.
type OpLog interface {
	// Append deduplicates on (docID, siteID, siteCounter) and returns the
	// accepted subset.
	Append(ctx context.Context, docID string, ops []crdt.Op) ([]crdt.Op, error)
	// Since returns every op the caller has not seen, per their version vector.
	Since(ctx context.Context, docID string, vv map[string]int64) ([]crdt.Op, error)
	LatestSnapshot(ctx context.Context, docID string) (url string, vv map[string]int64, err error)
	PutSnapshot(ctx context.Context, docID string, blob []byte, vv map[string]int64) (string, error)
	FetchSnapshot(ctx context.Context, url string) ([]byte, error)
}

type Authorizer interface {
	CanRead(ctx context.Context, userID, docID string) (bool, error)
	CanWrite(ctx context.Context, userID, docID string) (bool, error)
}

const (
	awarenessFlush        = 80 * time.Millisecond
	awarenessTTL          = 30 * time.Second
	snapshotEveryOps      = 5000
	snapshotInsteadOfOps  = 2000
)

type peer struct {
	conn        Connection
	awareness   *AwarenessState
	awarenessAt time.Time
}

// Room owns one document on one node. The hash ring decides which node that
// is; everything below assumes ownership has already been established.
type Room struct {
	DocID string

	mu               sync.Mutex
	doc              *crdt.RGA
	log              OpLog
	auth             Authorizer
	peers            map[string]*peer
	awarenessDirty   bool
	awarenessTimer   *time.Timer
	opsSinceSnapshot int
	draining         bool
}

// Open loads a room from the newest snapshot plus the tail of the op log.
func Open(ctx context.Context, docID string, log OpLog, auth Authorizer) (*Room, error) {
	site := "server:" + docID

	url, vv, err := log.LatestSnapshot(ctx, docID)
	if err != nil {
		return nil, err
	}

	var doc *crdt.RGA
	if url != "" {
		blob, err := log.FetchSnapshot(ctx, url)
		if err != nil {
			return nil, err
		}
		var snap crdt.Snapshot
		if err := json.Unmarshal(blob, &snap); err != nil {
			return nil, err
		}
		doc = crdt.FromSnapshot(site, snap)
	} else {
		doc = crdt.New(site)
		vv = map[string]int64{}
	}

	tail, err := log.Since(ctx, docID, vv)
	if err != nil {
		return nil, err
	}
	for _, op := range tail {
		_ = doc.ApplyRemote(op)
	}

	return &Room{
		DocID: docID, doc: doc, log: log, auth: auth,
		peers: map[string]*peer{},
	}, nil
}

func (r *Room) Handle(ctx context.Context, conn Connection, msg ClientMessage) error {
	r.mu.Lock()
	draining := r.draining
	r.mu.Unlock()
	if draining {
		conn.Send(ServerMessage{T: "error", Code: "MIGRATING", Message: "reconnect, this room is moving"})
		conn.Close("migrating")
		return nil
	}

	switch msg.T {
	case "hello":
		return r.onHello(ctx, conn, msg)
	case "push":
		return r.onPush(ctx, conn, msg.Ops)
	case "awareness":
		if msg.State != nil {
			r.onAwareness(conn, *msg.State)
		}
		return nil
	case "ping":
		conn.Send(ServerMessage{T: "pong"})
		return nil
	default:
		return fmt.Errorf("unknown message type %q", msg.T)
	}
}

func (r *Room) onHello(ctx context.Context, conn Connection, msg ClientMessage) error {
	ok, err := r.auth.CanRead(ctx, conn.UserID(), r.DocID)
	if err != nil {
		return err
	}
	if !ok {
		conn.Send(ServerMessage{T: "error", Code: "FORBIDDEN", Message: "no read access"})
		conn.Close("forbidden")
		return nil
	}

	r.mu.Lock()
	r.peers[conn.SiteID()] = &peer{conn: conn, awarenessAt: time.Now()}
	r.mu.Unlock()

	missing, err := r.log.Since(ctx, r.DocID, msg.VV)
	if err != nil {
		return err
	}

	// A client that is far behind gets a snapshot, not a wall of ops.
	if len(missing) > snapshotInsteadOfOps {
		url, vv, err := r.log.LatestSnapshot(ctx, r.DocID)
		if err == nil && url != "" {
			tail, err := r.log.Since(ctx, r.DocID, vv)
			if err != nil {
				return err
			}
			conn.Send(ServerMessage{T: "sync", SnapshotURL: url, Ops: tail, VV: r.versionVector()})
			r.broadcastAwarenessSoon()
			return nil
		}
	}

	conn.Send(ServerMessage{T: "sync", Ops: missing, VV: r.versionVector()})
	r.broadcastAwarenessSoon()
	return nil
}

func (r *Room) onPush(ctx context.Context, conn Connection, ops []crdt.Op) error {
	ok, err := r.auth.CanWrite(ctx, conn.UserID(), r.DocID)
	if err != nil {
		return err
	}
	if !ok {
		// The client applied these optimistically; tell it to roll back.
		conn.Send(ServerMessage{T: "error", Code: "READ_ONLY", Message: "rollback: no write access"})
		return nil
	}
	// Reject ops forged under someone else's site id.
	for _, op := range ops {
		if op.ID.Site != conn.SiteID() {
			conn.Send(ServerMessage{
				T: "error", Code: "BAD_SITE",
				Message: "op " + op.ID.Key() + " is not yours",
			})
			return nil
		}
	}

	// The unique index on (doc_id, site_id, site_counter) is what actually
	// guarantees dedupe. A replayed push writes nothing twice.
	accepted, err := r.log.Append(ctx, r.DocID, ops)
	if err != nil {
		return err
	}
	if len(accepted) == 0 {
		return nil
	}

	r.mu.Lock()
	for _, op := range accepted {
		_ = r.doc.ApplyRemote(op)
	}
	r.opsSinceSnapshot += len(accepted)
	needSnapshot := r.opsSinceSnapshot >= snapshotEveryOps
	if needSnapshot {
		r.opsSinceSnapshot = 0
	}
	targets := make([]Connection, 0, len(r.peers))
	for siteID, p := range r.peers {
		if siteID == conn.SiteID() {
			continue // the sender already applied them
		}
		targets = append(targets, p.conn)
	}
	r.mu.Unlock()

	for _, c := range targets {
		c.Send(ServerMessage{T: "ops", Ops: accepted})
	}

	if needSnapshot {
		// Snapshotting must never sit in the write path.
		go func() {
			if err := r.Snapshot(context.Background()); err != nil {
				fmt.Printf("snapshot failed for %s: %v\n", r.DocID, err)
			}
		}()
	}
	return nil
}

func (r *Room) onAwareness(conn Connection, state AwarenessState) {
	r.mu.Lock()
	if p, ok := r.peers[conn.SiteID()]; ok {
		p.awareness = &state
		p.awarenessAt = time.Now()
	}
	r.mu.Unlock()
	r.broadcastAwarenessSoon()
}

// broadcastAwarenessSoon coalesces: at most one broadcast per window, carrying
// only the latest state. Intermediate cursor positions are worthless.
func (r *Room) broadcastAwarenessSoon() {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.awarenessDirty = true
	if r.awarenessTimer != nil {
		return
	}
	r.awarenessTimer = time.AfterFunc(awarenessFlush, func() {
		r.mu.Lock()
		r.awarenessTimer = nil
		dirty := r.awarenessDirty
		r.awarenessDirty = false
		r.mu.Unlock()
		if dirty {
			r.flushAwareness()
		}
	})
}

func (r *Room) flushAwareness() {
	r.mu.Lock()
	now := time.Now()
	peers := make([]AwarenessPeer, 0, len(r.peers))
	for siteID, p := range r.peers {
		if p.awareness == nil || now.Sub(p.awarenessAt) > awarenessTTL {
			continue // expired, and never persisted anywhere
		}
		peers = append(peers, AwarenessPeer{SiteID: siteID, AwarenessState: *p.awareness})
	}
	targets := make([]Connection, 0, len(r.peers))
	for _, p := range r.peers {
		targets = append(targets, p.conn)
	}
	r.mu.Unlock()

	for _, c := range targets {
		c.Send(ServerMessage{T: "awareness", Peers: peers})
	}
}

func (r *Room) Disconnect(siteID string) {
	r.mu.Lock()
	delete(r.peers, siteID)
	r.mu.Unlock()
	r.broadcastAwarenessSoon()
}

func (r *Room) Snapshot(ctx context.Context) error {
	r.mu.Lock()
	blob, err := json.Marshal(r.doc.Snapshot())
	vv := r.doc.VersionVector()
	r.mu.Unlock()
	if err != nil {
		return err
	}
	_, err = r.log.PutSnapshot(ctx, r.DocID, blob, vv)
	return err
}

// Drain migrates the room: stop accepting work, flush a snapshot so the new
// owner starts cheap, then push every client to reconnect. Clients keep their
// local replica and unsent ops, so nothing is lost.
func (r *Room) Drain(ctx context.Context) error {
	r.mu.Lock()
	r.draining = true
	if r.awarenessTimer != nil {
		r.awarenessTimer.Stop()
		r.awarenessTimer = nil
	}
	targets := make([]Connection, 0, len(r.peers))
	for _, p := range r.peers {
		targets = append(targets, p.conn)
	}
	r.peers = map[string]*peer{}
	r.mu.Unlock()

	if err := r.Snapshot(ctx); err != nil {
		return err
	}
	for _, c := range targets {
		c.Send(ServerMessage{T: "error", Code: "MIGRATING", Message: "reconnect"})
		c.Close("migrating")
	}
	return nil
}

func (r *Room) versionVector() map[string]int64 {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.doc.VersionVector()
}

func (r *Room) MemberCount() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return len(r.peers)
}

func (r *Room) Text() string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.doc.String()
}