lww_map.go

package yin

import "reflect"

// lwwMapDeltaKind is the stable Delta.Kind tag for LWWMap deltas. Future
// codecs may route on this tag, but it is not itself a wire encoding.
const lwwMapDeltaKind = "lww-map"

// LWWMap is a string-keyed last-writer-wins map for values of type V.
//
// Each key stores its own update timestamp. Deletes are retained as internal
// tombstones so later merge and delta layers can distinguish an absent key from
// a removed key. Local writes are issued by replica and advance beyond the
// maximum counter already observed in the map's causal version.
//
// Generic values are treated as immutable once handed to the map and once
// returned to callers. The JSON object path is the explicit exception:
// LWWMap[json.RawMessage] clones raw JSON bytes when values cross public and
// sync boundaries so callers cannot mutate CRDT state without a timestamped
// write.
type LWWMap[V any] struct {
	replica ReplicaID
	entries map[string]lwwMapRecord[V]
	version VersionVector
}

// lwwMapRecord is the single internal representation for a timestamped map
// record, whether stored in a document, carried by a delta, or reconstructed
// from a snapshot. All value-ownership policy is funneled through the record
// constructors and clone/publicValue methods so json.RawMessage copies cannot
// be missed at individual API boundaries.
type lwwMapRecord[V any] struct {
	value     V
	timestamp Timestamp
	deleted   bool
}

var _ Document = (*LWWMap[int])(nil)

// NewLWWMap constructs an empty string-keyed LWW map whose future local writes
// are issued by replica. It does not validate replica; use a non-empty ID if
// deltas or snapshots from the map will be JSON-encoded.
func NewLWWMap[V any](replica ReplicaID) *LWWMap[V] {
	return &LWWMap[V]{
		replica: replica,
		entries: make(map[string]lwwMapRecord[V]),
	}
}

// Set applies a local write that stores value at key.
//
// For most V types, value is stored as-is and must be treated as immutable by
// the caller after Set returns. For V == json.RawMessage, Set clones the raw
// bytes before storing them.
func (m *LWWMap[V]) Set(key string, value V) {
	timestamp := m.nextTimestamp()
	m.storeRecord(key, newLWWMapRecord(value, timestamp, false))
	m.observeTimestamp(timestamp)
}

// Delete applies a local delete for key and retains an internal tombstone.
func (m *LWWMap[V]) Delete(key string) {
	timestamp := m.nextTimestamp()
	var zero V
	m.storeRecord(key, newLWWMapRecord(zero, timestamp, true))
	m.observeTimestamp(timestamp)
}

// Get returns the live value stored at key. Deleted or missing keys report ok
// false and return V's zero value. For V == json.RawMessage, Get returns a copy
// of the stored raw bytes.
func (m *LWWMap[V]) Get(key string) (value V, ok bool) {
	record, exists := m.entries[key]
	if !exists || record.deleted {
		var zero V
		return zero, false
	}
	return record.publicValue(), true
}

// Entries returns a copy of the map's live entries. Tombstones are omitted. For
// V == json.RawMessage, each returned value is a copy of the stored raw bytes.
func (m *LWWMap[V]) Entries() map[string]V {
	entries := make(map[string]V, len(m.entries))
	for key, record := range m.entries {
		if record.deleted {
			continue
		}
		entries[key] = record.publicValue()
	}
	return entries
}

// Merge incorporates entries and causal coverage from a compatible LWW map.
func (m *LWWMap[V]) Merge(other Document) (changed bool) {
	otherMap, ok := other.(*LWWMap[V])
	if !ok || otherMap == nil {
		return false
	}
	return m.applyRecords(otherMap.entries, otherMap.version)
}

// ExtractDelta returns the map records whose per-key timestamps are not already
// dominated by since, plus source causal coverage missing from since. A nil
// delta means since is caught up for the map's full causal version; otherwise
// the delta may carry version-only coverage for same-key writes that lost to
// retained records.
func (m *LWWMap[V]) ExtractDelta(since VersionVector) Delta {
	switch since.Compare(m.version) {
	case VersionVectorEqual, VersionVectorDominates:
		return nil
	}

	records := make(map[string]lwwMapRecord[V])
	for key, record := range m.entries {
		if lwwMapTimestampDominatedBy(record.timestamp, since) {
			continue
		}
		records[key] = record.clone()
	}
	return lwwMapDelta[V]{
		records: records,
		version: m.version.Since(since),
	}
}

// ApplyDelta incorporates compatible LWWMap delta records and causal coverage.
// Nil, incompatible, duplicate, and fully stale deltas are false no-ops.
// Causal version advancement includes safe version-only coverage carried by the
// delta for same-key writes that lost to retained records.
func (m *LWWMap[V]) ApplyDelta(d Delta) (changed bool) {
	if d == nil {
		return false
	}

	switch delta := d.(type) {
	case lwwMapDelta[V]:
		return m.applyDeltaRecords(delta.records, delta.version)
	case *lwwMapDelta[V]:
		if delta == nil {
			return false
		}
		return m.applyDeltaRecords(delta.records, delta.version)
	default:
		return false
	}
}

// Equal reports whether other is a compatible LWW map with the same entries,
// tombstones, per-key timestamps, and causal version coverage.
func (m *LWWMap[V]) Equal(other Document) bool {
	otherMap, ok := other.(*LWWMap[V])
	if !ok || otherMap == nil {
		return false
	}
	if !m.version.Equal(otherMap.version) || len(m.entries) != len(otherMap.entries) {
		return false
	}
	for key, record := range m.entries {
		otherRecord, exists := otherMap.entries[key]
		if !exists || record.deleted != otherRecord.deleted || record.timestamp != otherRecord.timestamp {
			return false
		}
		if !record.deleted && !reflect.DeepEqual(record.value, otherRecord.value) {
			return false
		}
	}
	return true
}

// Version returns a copy of the map's observed causal version, including
// tombstone updates.
func (m *LWWMap[V]) Version() VersionVector {
	return m.version.Clone()
}

func (m *LWWMap[V]) ensureEntries() {
	if m.entries == nil {
		m.entries = make(map[string]lwwMapRecord[V])
	}
}

func (m *LWWMap[V]) nextTimestamp() Timestamp {
	return nextLocalTimestamp(m.version, m.replica)
}

func (m *LWWMap[V]) observeTimestamp(timestamp Timestamp) (changed bool) {
	if timestamp.Counter() == 0 {
		return false
	}
	return m.applyVersion(lwwVersion(timestamp))
}

func (m *LWWMap[V]) applyRecords(records map[string]lwwMapRecord[V], version VersionVector) (changed bool) {
	for key, record := range records {
		if m.applyRecord(key, record) {
			changed = true
		}
	}
	if m.applyVersion(version) {
		changed = true
	}
	return changed
}

func (m *LWWMap[V]) applyRecord(key string, incoming lwwMapRecord[V]) (changed bool) {
	current, exists := m.entries[key]
	if exists && incoming.timestamp.Compare(current.timestamp) <= 0 {
		return false
	}

	m.storeRecord(key, incoming)
	return true
}

func (m *LWWMap[V]) storeRecord(key string, record lwwMapRecord[V]) {
	m.ensureEntries()
	m.entries[key] = record.clone()
}

func (m *LWWMap[V]) applyVersion(version VersionVector) (changed bool) {
	joined := m.version.Join(version)
	if joined.Equal(m.version) {
		return false
	}
	m.version = joined
	return true
}

func newLWWMapFromState[V any](replica ReplicaID, records map[string]lwwMapRecord[V], version VersionVector) *LWWMap[V] {
	return &LWWMap[V]{
		replica: replica,
		entries: cloneLWWMapRecords(records),
		version: version.Clone(),
	}
}

func cloneLWWMapRecords[V any](records map[string]lwwMapRecord[V]) map[string]lwwMapRecord[V] {
	clone := make(map[string]lwwMapRecord[V], len(records))
	for key, record := range records {
		clone[key] = record.clone()
	}
	return clone
}

func newLWWMapRecord[V any](value V, timestamp Timestamp, deleted bool) lwwMapRecord[V] {
	return lwwMapRecord[V]{
		value:     cloneLWWMapValue(value),
		timestamp: timestamp,
		deleted:   deleted,
	}
}

func (record lwwMapRecord[V]) clone() lwwMapRecord[V] {
	record.value = cloneLWWMapValue(record.value)
	return record
}

func (record lwwMapRecord[V]) publicValue() V {
	return cloneLWWMapValue(record.value)
}

func (m *LWWMap[V]) applyDeltaRecords(records map[string]lwwMapRecord[V], version VersionVector) (changed bool) {
	coverage := version.Clone()
	for key, record := range records {
		if record.timestamp.Counter() == 0 {
			continue
		}
		coverage = coverage.Join(lwwVersion(record.timestamp))
		if m.applyRecord(key, record) {
			changed = true
		}
	}

	if m.applyVersion(coverage) {
		changed = true
	}
	return changed
}

func lwwMapTimestampDominatedBy(timestamp Timestamp, version VersionVector) bool {
	return timestamp.Counter() == 0 || version.Get(timestamp.ReplicaID()) >= timestamp.Counter()
}

type lwwMapDelta[V any] struct {
	records map[string]lwwMapRecord[V]
	version VersionVector
}

func (d lwwMapDelta[V]) Kind() string {
	return lwwMapDeltaKind
}

func (d lwwMapDelta[V]) Version() VersionVector {
	return d.version.Clone()
}