#!/usr/bin/env python3
"""Deterministic state models, never a real Redis/replication/failure test."""
import argparse
import json
import itertools


class WatermarkCache:
    """One atomic operation per method; delivery/lifetime are explicit inputs."""
    def __init__(self):
        self.floor = 0
        self.value = None

    def invalidate(self, version):
        self.floor = max(self.floor, version)
        if self.value is not None and self.value[0] < self.floor:
            self.value = None

    def fill(self, version, payload):
        if version < self.floor or (self.value is not None and version < self.value[0]):
            return False
        if self.value is not None and version == self.value[0] and payload != self.value[1]:
            return False  # fixed projection: equal version cannot change payload
        self.value = (version, payload)
        return True

    def read(self):
        if self.value is None or self.value[0] < self.floor:
            return None
        return self.value[0]


def cache_model():
    database = 1
    old_snapshot = database
    database = 2
    cache = None  # successful writer deletion
    cache = old_snapshot  # delayed reader refills after commit
    guarded = WatermarkCache()
    guarded.invalidate(database)
    accepted = guarded.fill(old_snapshot, 'old')
    guarded.fill(database, 'new')
    current = guarded.read()
    guarded.value = (1, 'injected stale value')  # bad data must fail the read check
    stale_read = guarded.read()
    pending = WatermarkCache()
    pending.fill(1, 'old')  # DB2 committed, invalidation event has not arrived
    lost = WatermarkCache()
    lost.invalidate(2)
    lost.floor = 0  # eviction / expiry / failover loses F2
    lost.fill(old_snapshot, 'old')
    deleted = WatermarkCache()
    deleted.invalidate(3)  # delete carries a version/tombstone
    normal = cache_interleavings()
    partial = WatermarkCache()
    partial.invalidate(1)
    partial.fill(1, 'old')
    observed_floor = partial.floor
    partial.floor = 2  # Lua SET F2 succeeds, next command errors before deleting C1
    split_read = partial.value[0] if partial.value[0] >= observed_floor else None
    equal = WatermarkCache()
    first = equal.fill(2, 'canonical')
    duplicate = equal.fill(2, 'canonical')
    conflict = equal.fill(2, 'different')
    return {
        'kind': 'deterministic-model', 'stale_refill': cache,
        'stale_until_ms_after_commit': 600000 + 1000,
        'stale_fill_accepted': accepted, 'current_fill_version': current,
        'stale_read': stale_read, 'pending_event_read': pending.read(),
        'lost_watermark_fill_version': lost.read(),
        'deleted_object_refill_accepted': deleted.fill(2, 'resurrect'),
        'normal_interleavings': normal,
        'partial_script_failure': {'floor': partial.floor, 'cache': partial.value[0],
                                   'split_read': split_read, 'guarded_read': partial.read()},
        'same_version': {'first': first, 'duplicate': duplicate, 'conflict': conflict, 'payload': equal.value[1]},
    }


def cache_interleavings():
    """All permutations of one split read plus atomic fill/invalidate calls.

    Initial F1/C1; immutable payload per version. The only ordering requirement
    is GET F before GET C. No error, eviction or bypass writes are permitted.
    """
    histories, violations = 0, []
    for events in itertools.permutations(('get-F', 'get-C', 'invalidate-2', 'fill-1', 'fill-2')):
        if events.index('get-F') > events.index('get-C'):
            continue
        histories += 1
        cache = WatermarkCache()
        cache.invalidate(1)
        cache.fill(1, 'v1')
        read_floor = None
        for event in events:
            if event == 'get-F':
                read_floor = cache.floor
            elif event == 'get-C':
                returned = cache.value[0] if cache.value and cache.value[0] >= read_floor else None
                if returned is not None and returned < cache.floor:
                    violations.append({'events': events, 'at': event})
            elif event == 'invalidate-2':
                cache.invalidate(2)
            elif event == 'fill-1':
                cache.fill(1, 'v1')
            else:
                cache.fill(2, 'v2')
            if cache.value and cache.value[0] < cache.floor:
                violations.append({'events': events, 'at': event})
    return {'histories': histories, 'violations': len(violations), 'counterexample': next(iter(violations), None)}


def legacy_models():
    # Preserve the earlier shared projection counterexample for its owning topic.
    projection = {'version': 2, 'value': 'new'}
    event = {'version': 1, 'value': 'old'}
    if event['version'] > projection['version']:
        projection = event
    assert projection['value'] == 'new'
    needed = max(5 // 2 + 1, 2)
    assert needed == 3
    return {'projection_value': projection['value'], 'sentinel_5_quorum_2_needed': needed}


def recovery_model():
    def partial(history, offset, backlog=True):
        identity = history == 'current' or (history == 'second' and offset <= 150)
        return identity and backlog and 100 <= offset <= 201

    def authorized(votes, quorum):
        return votes >= 5 // 2 + 1 and votes >= quorum

    return {
        'kind': 'deterministic-model',
        'psync': {
            'current': partial('current', 150), 'wrong_history': partial('other', 150),
            'second_at_cutoff': partial('second', 150), 'second_after_cutoff': partial('second', 151),
            'before_backlog': partial('current', 99), 'at_end_plus_one': partial('current', 201),
            'after_end_plus_one': partial('current', 202), 'missing_backlog': partial('current', 150, False),
        },
        'sentinel': {
            'two_of_five_quorum2': authorized(2, 2), 'three_of_five_quorum2': authorized(3, 2),
            'three_of_five_quorum4': authorized(3, 4), 'four_of_five_quorum4': authorized(4, 4),
        },
    }


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--json', action='store_true', help='machine-readable output (also the default)')
    parser.add_argument('mode', nargs='?', choices=('cache', 'recovery', 'all'), default='all')
    args = parser.parse_args()
    results = {'cache': cache_model(), 'recovery': recovery_model(), 'legacy': legacy_models()}
    result = results[args.mode] if args.mode != 'all' else results
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()