#!/usr/bin/env python3
"""Fail-closed verification of approved front zero-date/restaurant-permission changes.

Reads only private source/final snapshots and explicit approval JSON. Does not
connect to a database. No customer row values are printed in reports.
"""
import argparse
from collections import Counter
import copy
import gzip
import hashlib
import json
from pathlib import Path
import re

ZERO_FIELDS = {
    ('ydy_menu', 'create_time'), ('ydy_metabolism_pal', 'turnover_time'),
    ('ydy_station', 'create_time'), ('ydy_user', 'useful_start'),
    ('ydy_user', 'useful_end'),
}
NULLABLE_FIELDS = ZERO_FIELDS - {('ydy_user', 'useful_start'), ('ydy_user', 'useful_end')}
SAFE_NAME = re.compile(r'^[A-Za-z0-9_]+$')


class InvalidEvidence(ValueError):
    pass


def require(condition, message):
    if not condition:
        raise InvalidEvidence(message)


def digest(file):
    return hashlib.sha256(Path(file).read_bytes()).hexdigest()


def load_baseline(file):
    file = Path(file)
    obj = json.loads(file.read_text(encoding='utf-8'))
    require(set(obj['metadata']) == set(obj['counts']), 'Baseline table/count keys mismatch')
    require(obj['tables'] == len(obj['metadata']), 'Baseline table count mismatch')
    require(obj['rows'] == sum(obj['counts'].values()), 'Baseline row total mismatch')
    return obj


def ddl_info(folder, table, columns):
    ddl = (folder / (table + '.ddl.txt')).read_text(encoding='utf-8')
    pk = re.search(r'PRIMARY KEY\s*\(([^)]+)\)', ddl)
    require(pk is not None, 'No primary key in ' + table)
    names = re.findall(r'`([^`]+)`', pk.group(1))
    require(names and set(names) <= {c['name'] for c in columns}, 'Invalid primary key in ' + table)
    seed = re.search(r'\bAUTO_INCREMENT=(\d+)', ddl)
    has_identity = any('auto_increment' in c.get('extra', '') for c in columns)
    counter = int(seed.group(1)) if seed else 1 if has_identity else None
    return ddl, tuple(names), counter


def load_table(folder, table, columns, primary_key):
    names = [c['name'] for c in columns]
    require(len(set(names)) == len(names), 'Duplicate metadata column in ' + table)
    records = {}
    with gzip.open(folder / (table + '.hex.jsonl.gz'), 'rt', encoding='utf-8') as stream:
        for line in stream:
            raw = json.loads(line)
            require(isinstance(raw, list) and len(raw) == len(names), 'Invalid row shape in ' + table)
            values = []
            for v in raw:
                if v is None:
                    values.append(None)
                else:
                    require(isinstance(v, str) and re.fullmatch(r'(?:[0-9a-fA-F]{2})*', v) is not None,
                            'Invalid HEX in ' + table)
                    values.append(bytes.fromhex(v).decode('utf-8'))
            row = dict(zip(names, values))
            key = tuple(row[c] for c in primary_key)
            require(all(v is not None for v in key), 'NULL primary key in ' + table)
            require(key not in records, 'Duplicate primary key in ' + table)
            records[key] = row
    return records


def canonical_ddl(text, table, nullable_columns):
    # Ignore only the table AUTO_INCREMENT option, checked independently below.
    lines = re.sub(r'\bAUTO_INCREMENT=\d+\s*', '', text).splitlines()
    for i, line in enumerate(lines):
        match = re.match(r'\s*`([^`]+)`\s+', line)
        if match and (table, match.group(1)) in nullable_columns:
            # Authorized fields have no explicit non-NULL default. Canonicalize
            # exactly the nullability phrase, not arbitrary types/comments.
            lines[i] = re.sub(r'\b(?:NOT NULL|DEFAULT NULL)\b', '<APPROVED_NULLABILITY>', line, count=1)
    return '\n'.join(lines)


def verify(before_file, after_file, approval_file):
    before_file, after_file = Path(before_file), Path(after_file)
    before = load_baseline(before_file)
    after = load_baseline(after_file)
    plan = json.loads(Path(approval_file).read_text(encoding='utf-8'))
    require(before['db'] == plan['source_database'] and after['db'] == plan['target_database'],
            'Approved database mismatch')
    require(set(before['metadata']) == set(after['metadata']), 'Table set changed')
    require(all(SAFE_NAME.fullmatch(t) for t in before['metadata']), 'Unsafe table name')

    nullable = plan['nullable_changes']
    require(isinstance(nullable, list) and len(nullable) == 3, 'Expected exactly three nullable changes')
    keys = [(x['table'], x['column']) for x in nullable]
    require(set(keys) == NULLABLE_FIELDS and len(set(keys)) == 3, 'Unexpected nullable field scope')
    require(all(x['before'] == 'NO' and x['after'] == 'YES' for x in nullable), 'Invalid nullability transition')
    expected_metadata = copy.deepcopy(before['metadata'])
    for table, column in keys:
        matches = [c for c in expected_metadata[table] if c['name'] == column]
        require(len(matches) == 1 and matches[0]['nullable'] == 'NO', 'Source nullability differs from approval')
        matches[0]['nullable'] = 'YES'
    require(expected_metadata == after['metadata'], 'Unexpected metadata/column/default change')

    cells_by_table = {}
    seen_cells = set()
    for change in plan['zero_date_cells']:
        table, column = change['table'], change['column']
        require((table, column) in ZERO_FIELDS, 'Unapproved zero-date field scope')
        require(change['before'] in ('0000-00-00', '0000-00-00 00:00:00') and change['after'] is None,
                'Only zero-date to NULL is approved')
        require(isinstance(change['pk'], dict) and all(isinstance(v, str) for v in change['pk'].values()),
                'Invalid approved primary key')
        signature = (table, tuple(sorted(change['pk'].items())), column)
        require(signature not in seen_cells, 'Duplicate approved cell')
        seen_cells.add(signature)
        cells_by_table.setdefault(table, []).append(change)

    permission = plan['permission_insert']
    require(permission['table'] == 'ydy_user_data' and isinstance(permission['row'], dict),
            'Permission insert table mismatch')
    new_permission = permission['row']
    require(set(new_permission) == {c['name'] for c in before['metadata']['ydy_user_data']},
            'Permission insert must specify every field')
    require(all(v is None or isinstance(v, str) for v in new_permission.values()),
            'Permission values must be string/null')
    require(new_permission['type'] == '3' and new_permission['staff_uuid_str'] is None,
            'Only exact restaurant grant without staff wildcard is approved')
    increment = plan['auto_increment_change']
    require(increment['table'] == 'ydy_user_data', 'Wrong counter exception table')
    require(str(increment['before']).isdigit() and str(increment['after']).isdigit(), 'Invalid counter values')
    require(int(increment['after']) == int(increment['before']) + 1, 'Permission counter must advance exactly one')
    require(new_permission['id'] == str(increment['before']), 'Permission ID must equal original next counter')

    changed_cells = 0
    total_before = total_after = 0
    details = []
    reference_rows = {}
    for table, columns in before['metadata'].items():
        ddl_before, pk, seed_before = ddl_info(before_file.parent, table, columns)
        ddl_after, pk_after, seed_after = ddl_info(after_file.parent, table, after['metadata'][table])
        require(pk == pk_after, 'Primary key changed in ' + table)
        require(canonical_ddl(ddl_before, table, NULLABLE_FIELDS) ==
                canonical_ddl(ddl_after, table, NULLABLE_FIELDS), 'Unexpected DDL/index/table-option change in ' + table)
        if table == 'ydy_user_data':
            require(seed_before == int(increment['before']) and seed_after == int(increment['after']),
                    'Permission AUTO_INCREMENT differs from approval')
        else:
            require(seed_before == seed_after, 'Unapproved AUTO_INCREMENT change in ' + table)
        original = load_table(before_file.parent, table, columns, pk)
        actual = load_table(after_file.parent, table, after['metadata'][table], pk)
        require(len(original) == before['counts'][table] and len(actual) == after['counts'][table],
                'Row count manifest mismatch in ' + table)
        total_before += len(original)
        total_after += len(actual)
        expected = copy.deepcopy(original)
        for change in cells_by_table.get(table, []):
            require(set(change['pk']) == set(pk), 'Approved key shape mismatch in ' + table)
            key = tuple(change['pk'][name] for name in pk)
            require(key in expected and expected[key][change['column']] == change['before'],
                    'Approved original cell does not match in ' + table)
            expected[key][change['column']] = None
            changed_cells += 1
        if table == 'ydy_user_data':
            key = tuple(new_permission[name] for name in pk)
            require(key not in expected, 'Approved permission insert overwrites existing row')
            require(not any(r['user_uuid'] == new_permission['user_uuid'] and
                            r['data_uuid'] == new_permission['data_uuid'] and r['type'] == '3'
                            for r in original.values()), 'Equivalent restaurant permission already exists')
            expected[key] = new_permission
        require(expected == actual, 'Unapproved row/field/permission change in ' + table)
        for row in actual.values():
            for t, column in ZERO_FIELDS:
                if t == table:
                    require(row[column] not in ('0000-00-00', '0000-00-00 00:00:00'),
                            'Unnormalized zero date remains in approved scope')
        if table in ('ydy_user', 'ydy_restaurants'):
            reference_rows[table] = original
        details.append({'table': table, 'before_rows': len(original), 'after_rows': len(actual),
                        'approved_cell_changes': len(cells_by_table.get(table, [])), 'equal_with_approval': True})

    owners = [r for r in reference_rows['ydy_user'].values() if r['uuid'] == new_permission['user_uuid']]
    restaurants = [r for r in reference_rows['ydy_restaurants'].values() if r['uuid'] == new_permission['data_uuid']]
    require(len(owners) == 1 and len(restaurants) == 1, 'Permission owner or restaurant is not unique in its own source')
    require(changed_cells == len(plan['zero_date_cells']), 'Approved cell count mismatch')
    require(total_after == total_before + 1, 'Expected only one inserted row')
    return {'verified': True, 'source_database': before['db'], 'target_database': after['db'],
            'tables': len(details), 'rows_before': total_before, 'rows_after': total_after,
            'approved_zero_date_cells': changed_cells, 'approved_nullable_columns': 3,
            'approved_permission_rows_inserted': 1, 'unapproved_changes': 0,
            'evidence_sha256': {'before_baseline': digest(before_file), 'after_baseline': digest(after_file),
                                'approval': digest(approval_file)}, 'table_results': details}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    for name in ('before', 'after', 'approval', 'output'):
        parser.add_argument('--' + name, required=True)
    args = parser.parse_args()
    try:
        report = verify(args.before, args.after, args.approval)
    except Exception as error:
        report = {'verified': False, 'error': str(error) if isinstance(error, InvalidEvidence) else type(error).__name__}
    Path(args.output).write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps({k: v for k, v in report.items() if k != 'table_results'}, ensure_ascii=False))
    return 0 if report['verified'] else 1


if __name__ == '__main__':
    raise SystemExit(main())