#!/usr/bin/env python3
"""Bounded real Redis cache protocol assertions, never a DB/CDC/failover test.

Download run_local.py into the same directory. Only three UUID-owned keys are
written; finally deletes exactly those keys. No global configuration changes.
Lua below is teaching code, not Redis engine source.
Versions here are fixed small positive integers 1/2/3, exactly representable
by Lua doubles. Do not use tonumber for arbitrary BIGINT version comparison.
"""
import argparse
import json
import uuid

from run_local import Client, Experiment, RedisError, info


INVALIDATE = '''
local oldF = tonumber(redis.call('GET', KEYS[1]) or '0')
local oldC = tonumber(redis.call('HGET', KEYS[2], 'v') or '0')
local nextF = math.max(oldF, tonumber(ARGV[1]))
redis.call('SET', KEYS[1], nextF)
if oldC < nextF then redis.call('DEL', KEYS[2]) end
return nextF
'''
FILL = '''
local f = tonumber(redis.call('GET', KEYS[1]) or '0')
local c = tonumber(redis.call('HGET', KEYS[2], 'v') or '0')
local v = tonumber(ARGV[1])
if v < f or v < c then return 'STALE' end
if v == c then
  if redis.call('HGET', KEYS[2], 'p') ~= ARGV[2] then return 'CONFLICT' end
  return 'DUPLICATE'
end
redis.call('HSET', KEYS[2], 'v', v, 'p', ARGV[2])
redis.call('EXPIRE', KEYS[2], 120)
return 'APPLIED'
'''
READ = '''
local f = tonumber(redis.call('GET', KEYS[1]) or '0')
local c = tonumber(redis.call('HGET', KEYS[2], 'v') or '0')
if c == 0 or c < f then return false end
return c
'''


def run(client, exp):
    floor, cache, auxiliary = [exp.key(name, empty=True) for name in ('floor', 'cache', 'bad-auxiliary')]
    keys = (floor, cache)
    invalidate = lambda v: client.call('EVAL', INVALIDATE, 2, *keys, v)
    fill = lambda v, p: client.call('EVAL', FILL, 2, *keys, v, p)
    read = lambda: client.call('EVAL', READ, 2, *keys)
    invalidate(1)
    fill(1, 'canonical-v1')
    first_floor = int(client.call('GET', floor))
    invalidate(2)  # a normal atomic invalidate between the two reads
    normal_split_value = client.call('HGET', cache, 'v')
    assert first_floor == 1 and normal_split_value is None
    old = fill(1, 'canonical-v1')
    new = fill(2, 'canonical-v2')
    duplicate = fill(2, 'canonical-v2')
    conflict = fill(2, 'different-v2')
    current = read()
    assert [old, new, duplicate, conflict, current] == ['STALE', 'APPLIED', 'DUPLICATE', 'CONFLICT', 2]
    invalidate(3)  # deletion/tombstone event
    deleted = fill(2, 'canonical-v2')
    assert deleted == 'STALE'

    client.call('DEL', floor, cache)  # exact own keys, prepare the fault history
    invalidate(1)
    fill(1, 'canonical-v1')
    split_floor = int(client.call('GET', floor))
    client.call('SET', auxiliary, 'not-an-integer', 'EX', 120)
    try:
        client.call('EVAL', "redis.call('SET',KEYS[1],2); redis.call('INCR',KEYS[3]); redis.call('DEL',KEYS[2])", 3, floor, cache, auxiliary)
        raise AssertionError('injected script must error after setting floor')
    except RedisError as error:
        error_message = str(error)
    stale = int(client.call('HGET', cache, 'v'))
    split_read = stale if stale >= split_floor else None
    guarded_read = read()
    installed = int(client.call('GET', floor))
    assert [installed, stale, split_read, guarded_read] == [2, 1, 1, None]
    client.call('DEL', floor, cache)  # explicit loss/rebuild fault injection
    fill(1, 'canonical-v1')
    lost_floor_read = read()
    assert lost_floor_read == 1
    return {'normal_split': {'observed_floor': first_floor, 'cache_after_invalidate': normal_split_value},
            'fills': [old, new, duplicate, conflict], 'current_read': current,
            'tombstone_reject': deleted,
            'partial_script_failure': {'error': error_message, 'floor': installed, 'cache': stale,
                                       'split_read': split_read, 'guarded_read': guarded_read},
            'lost_floor_read': lost_floor_read}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--host', default='127.0.0.1')
    parser.add_argument('--port', type=int, default=6379)
    args = parser.parse_args()
    client = Client(args.host, args.port)
    exp = Experiment(client)
    exp.prefix = 'study_t20_' + uuid.uuid4().hex + ':'
    try:
        server, memory, replication = [info(client, section) for section in ('server', 'memory', 'replication')]
        if replication['role'] != 'master' or server['redis_mode'] != 'standalone':
            raise RuntimeError('bounded local lab requires a writable standalone master')
        used, maximum = int(memory['used_memory']), int(memory['maxmemory'])
        if maximum and maximum - used < 16 * 1024 * 1024:
            raise RuntimeError('less than 16 MiB maxmemory headroom; refusing all writes')
        result = {'kind': 'real-redis-standalone-assertions',
                  'runtime_version': server['redis_version'],
                  'teaching_source_baseline': '7.2.5', 'namespace': exp.prefix,
                  'memory_guard': {'used_memory': used, 'maxmemory': maximum, 'required_capped_headroom': 16777216},
                  'version_domain': 'fixed positive integers 1/2/3; not arbitrary BIGINT',
                  'assertions': run(client, exp)}
    finally:
        try:
            cleanup = exp.cleanup()
        finally:
            client.close()
    result['cleanup'] = cleanup
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()