"""Independent exact-value migration verifier; never emits business rows.

Wire: NDJSON {"table": "lowercase_table", "row": [str | None, ...]}.
Only native MySQL JSON, exact numeric columns and DOUBLE spelling are
canonicalized. JSON-looking TEXT, Unicode and time strings remain exact.
"""
import argparse
import collections
from decimal import Decimal, InvalidOperation, localcontext
import gzip
import hashlib
import json
from pathlib import Path
import re
import sys


VERSION = 'foodsafe-dm-values-v1'
INTEGER_TYPES = {'tinyint', 'smallint', 'mediumint', 'int', 'integer', 'bigint'}
NUMBER_LITERAL = re.compile(r'^[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?$')


class ContractError(Exception):
    pass


def packed(value):
    return json.dumps(value, ensure_ascii=False, separators=(',', ':'),
                      allow_nan=False).encode('utf8')


def exact_decimal(value):
    if not NUMBER_LITERAL.fullmatch(value):
        raise ContractError('Invalid numeric representation')
    try:
        result = Decimal(value)
    except InvalidOperation:
        raise ContractError('Invalid numeric representation') from None
    if not result.is_finite():
        raise ContractError('Nonfinite numeric value')
    return result


def number_key(value):
    """No Decimal.normalize(): its context can round >28-digit values."""
    if not value.is_finite():
        raise ContractError('Nonfinite JSON number')
    if value.is_zero():
        return ['number', '0', '0']
    sign, digits, exponent = value.as_tuple()
    digits = list(digits)
    while digits[-1] == 0:
        digits.pop()
        exponent += 1
    coefficient = ('-' if sign else '') + ''.join(str(x) for x in digits)
    return ['number', coefficient, str(exponent)]


def object_pairs(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ContractError('Duplicate native JSON object key')
        result[key] = value
    return result


def json_node(value):
    if value is None:
        return ['null']
    if isinstance(value, bool):
        return ['bool', value]
    if isinstance(value, Decimal):
        return number_key(value)
    if isinstance(value, str):
        return ['string', value]
    if isinstance(value, list):
        return ['array', [json_node(x) for x in value]]
    if isinstance(value, dict):
        return ['object', [[key, json_node(value[key])] for key in sorted(value)]]
    raise ContractError('Unsupported native JSON value')


def native_json(value):
    def reject_constant(_):
        raise ContractError('Nonfinite native JSON number')
    try:
        result = json.loads(value, parse_float=Decimal, parse_int=Decimal,
                            object_pairs_hook=object_pairs,
                            parse_constant=reject_constant)
        return json_node(result)
    except (ValueError, InvalidOperation, RecursionError):
        raise ContractError('Invalid native JSON value') from None


def scaled_decimal(value, column):
    number = exact_decimal(value)
    scale = int(column.get('NUMERIC_SCALE') or 0)
    precision = column.get('NUMERIC_PRECISION')
    if scale < 0 or scale > 1000:
        raise ContractError('Invalid decimal metadata')
    if precision is not None and not number.is_zero():
        if max(number.adjusted() + 1, 0) > int(precision) - scale:
            raise ContractError('Decimal exceeds source precision')
    # Reject over-scale values instead of silently rounding them to source scale.
    with localcontext() as context:
        context.prec = max(len(number.as_tuple().digits), int(precision or 1),
                           scale + 1) + scale + 4
        try:
            quantized = number.quantize(Decimal(1).scaleb(-scale))
        except InvalidOperation:
            raise ContractError('Decimal cannot fit source scale') from None
    if quantized != number:
        raise ContractError('Decimal exceeds source scale')
    if quantized.is_zero():
        quantized = abs(quantized)
    return format(quantized, '.' + str(scale) + 'f')


class Contract:
    def __init__(self, metadata):
        names = [t['TABLE_NAME'] for t in metadata['tables']]
        if len(names) != len(set(names)) or any(n != n.lower() for n in names):
            raise ContractError('Invalid table metadata')
        self.columns = {name: [] for name in sorted(names)}
        for column in metadata['columns']:
            name = column['TABLE_NAME']
            if name not in self.columns:
                raise ContractError('Column references unknown table')
            self.columns[name].append(column)
        for columns in self.columns.values():
            columns.sort(key=lambda c: c['ORDINAL_POSITION'])
        description = [[table, [[c['COLUMN_NAME'], c['DATA_TYPE'],
                                c.get('NUMERIC_PRECISION'), c.get('NUMERIC_SCALE'),
                                c.get('DATETIME_PRECISION')]
                               for c in columns]]
                       for table, columns in self.columns.items()]
        self.metadata_sha256 = hashlib.sha256(packed(description)).hexdigest()

    def canonical_row(self, table, row):
        if table not in self.columns:
            raise ContractError('Unknown table in verification stream')
        columns = self.columns[table]
        if not isinstance(row, list) or len(row) != len(columns):
            raise ContractError('Invalid row width for table ' + table)
        result = []
        for column, value in zip(columns, row):
            if value is None:
                result.append(['sql-null'])
                continue
            if not isinstance(value, str):
                raise ContractError('Wire values must be string or SQL NULL')
            kind = column['DATA_TYPE'].lower()
            try:
                if kind == 'json':
                    item = ['json', native_json(value)]
                elif kind in INTEGER_TYPES:
                    if not re.fullmatch(r'[+-]?\d+', value):
                        raise ContractError('Invalid integer representation')
                    item = ['integer', str(int(value))]
                elif kind in {'decimal', 'numeric'}:
                    item = ['decimal', scaled_decimal(value, column)]
                elif kind in {'double', 'float', 'real'}:
                    # Exact decimal spelling equivalence only, no epsilon and
                    # no Python float. 0.1 differs from 0.10000000000000001.
                    item = ['floating-spelling', number_key(exact_decimal(value))]
                else:
                    item = ['exact', value]
            except ContractError as error:
                raise ContractError(str(error) + ' at ' + table + '.' +
                                    column['COLUMN_NAME']) from None
            result.append(item)
        # Nested JSON arrays delimit columns, types, NULL and arbitrary text;
        # a simple string concatenation would have ambiguous boundaries.
        return packed(result)

    def summarize(self, records):
        hashes = {name: [] for name in self.columns}
        for record in records:
            if not isinstance(record, dict) or set(record) != {'table', 'row'}:
                raise ContractError('Invalid verification record envelope')
            table = record['table']
            if not isinstance(table, str):
                raise ContractError('Invalid verification table name')
            raw = self.canonical_row(table, record['row'])
            hashes[table].append(hashlib.sha256(raw).digest())
        tables = {}
        for table, row_hashes in hashes.items():
            row_hashes.sort()
            digest = hashlib.sha256()
            for row_hash in row_hashes:
                digest.update(row_hash)
            tables[table] = {'count': len(row_hashes),
                             'multiset_sha256': digest.hexdigest()}
        return {'contract_version': VERSION,
                'metadata_sha256': self.metadata_sha256,
                'total_rows': sum(t['count'] for t in tables.values()),
                'tables': tables}


def compare_manifests(expected, actual):
    differences = {}
    for name in ['contract_version', 'metadata_sha256']:
        if expected.get(name) != actual.get(name):
            differences['__' + name] = {'mismatch': True}
    expected_tables = expected.get('tables', {})
    actual_tables = actual.get('tables', {})
    for table in sorted(set(expected_tables) | set(actual_tables)):
        if expected_tables.get(table) != actual_tables.get(table):
            differences[table] = {'expected': expected_tables.get(table),
                                  'actual': actual_tables.get(table)}
    if expected.get('total_rows') != actual.get('total_rows'):
        differences['__total_rows'] = {'expected': expected.get('total_rows'),
                                       'actual': actual.get('total_rows')}
    return differences


def read_ndjson(stream):
    for line in stream:
        if not line.strip():
            raise ContractError('Blank line in verification stream')
        try:
            yield json.loads(line)
        except (ValueError, RecursionError):
            raise ContractError('Invalid verification stream JSON') from None


def source_records(root, metadata):
    for table in sorted(t['TABLE_NAME'] for t in metadata['tables']):
        path = root / 'private/upgraded-export' / (table + '.jsonl.gz')
        if not path.exists():
            if int(metadata['counts'][table]):
                raise ContractError('Missing nonempty source export ' + table)
            continue
        with gzip.open(path, 'rt', encoding='utf8') as stream:
            for row in read_ndjson(stream):
                yield {'table': table, 'row': row}


def save_manifest(path, manifest):
    with path.open('x', encoding='utf8') as stream:
        json.dump(manifest, stream, ensure_ascii=False, indent=2, allow_nan=False)
        stream.write('\n')
    path.chmod(0o600)


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--mode', choices=['source', 'target'], required=True)
    parser.add_argument('--root', type=Path, default=Path(__file__).resolve().parents[1])
    parser.add_argument('--output', type=Path)
    parser.add_argument('--expected', type=Path)
    args = parser.parse_args(argv)
    try:
        metadata = json.loads((args.root / 'private/mysql-upgraded-baseline.json').read_text())
        contract = Contract(metadata)
        expected_path = args.expected or args.root / 'private/expected-normalized-manifest.json'
        if args.mode == 'source':
            manifest = contract.summarize(source_records(args.root, metadata))
            for table, count in metadata['counts'].items():
                if manifest['tables'][table]['count'] != int(count):
                    raise ContractError('Source baseline row count mismatch ' + table)
            output = args.output or expected_path
            save_manifest(output, manifest)
            print(json.dumps({'source_manifest_complete': True,
                              'tables': len(manifest['tables']),
                              'total_rows': manifest['total_rows'],
                              'contract_version': VERSION}))
            return 0
        actual = contract.summarize(read_ndjson(sys.stdin))
        expected = json.loads(expected_path.read_text())
        differences = compare_manifests(expected, actual)
        actual['comparison'] = {'equal': not differences, 'differences': differences}
        output = args.output or args.root / 'private/actual-normalized-manifest.json'
        save_manifest(output, actual)
        print(json.dumps({'equal': not differences, 'tables': len(actual['tables']),
                          'total_rows': actual['total_rows'],
                          'different_tables_or_contract_fields': sorted(differences)}))
        return 0 if not differences else 1
    except (ContractError, OSError, ValueError, KeyError, UnicodeError, EOFError):
        # Tracebacks and parser messages can embed customer values. Fail closed
        # without printing row content or exception arguments.
        print('Verification failed: invalid metadata, stream, or value contract; no business values logged.',
              file=sys.stderr)
        return 2


if __name__ == '__main__':
    sys.exit(main())