#!/usr/bin/env python3
"""SQL API parity probe: the planner's contract through every host's own SQL
binding. tools/check-sqlapi.sh diffs the reports of the hosts that carry the
SQL layer (the JS bundles do not; they print nothing and are left out).
Three programs -- one the planner pushes down whole, one it splits into a SQL
prefix and an in-memory continuation, one it keeps in memory -- and the same
nine questions about each plan: its classification, dialect, statement, prefix
and continuation presence, the continuation's dependencies, its source
variable, the physical source tables and the selected member. The probe NAMES
are the contract and the VALUES are compared; each host spells its accessors
its own way (SEL-0044).
"""
import sys

import _harness  # noqa: F401 - first: it makes `sel` importable (the probe)

import sel                                                          # noqa: E402
from sel.sql import Binding, Sql                                    # noqa: E402

out = []
_n = [0]


def say(name, value):
    _n[0] += 1
    out.append(f'{_n[0]:02d} {name} = {value}')


def b(x):
    return 'true' if x else 'false'


BINDINGS = {
    'ORDERS': Binding.relation('orders', 'o', {
        'ID': Binding.column('id', 'o', 'NUM'), 'CUSTOMER_ID': Binding.column('customer_id', 'o', 'NUM'),
        'AMOUNT': Binding.column('amount', 'o', 'NUM'), 'NAME': Binding.column('name', 'o', 'TEXT')}, None, None),
    'CUSTOMERS': Binding.relation('customers', 'c', {
        'ID': Binding.column('id', 'c', 'NUM'), 'NAME': Binding.column('name', 'c', 'TEXT')}, None, None),
}


def probe(label, source):
    plan = Sql.plan_hybrid(sel.compile(source), 'mariadb', BINDINGS)
    say(f'plan.{label}.kind', plan.kind)
    say(f'plan.{label}.dialect', plan.dialect or '-')
    say(f'plan.{label}.statement', plan.sql_statement.as_statement() if plan.sql_statement is not None else '-')
    say(f'plan.{label}.prefix.present', b(plan.sql_prefix_ast is not None))
    say(f'plan.{label}.continuation.present', b(plan.continuation_ast is not None))
    cont = plan.continuation_program
    say(f'plan.{label}.continuation.deps', ' '.join(cont.dependencies()) if cont is not None else '-')
    say(f'plan.{label}.source.var', plan.continuation_source_var)
    say(f'plan.{label}.tables', ','.join(plan.source_tables))
    say(f'plan.{label}.selected.member', 'present' if plan.selected_member is not None else '-')


def fragment_probe(label, dialect, source):
    f = Sql.translate(sel.compile(source), dialect, BINDINGS)
    say(f'fragment.{label}.kind', f.kind)
    say(f'fragment.{label}.canonical', b(f.canonical))
    say(f'fragment.{label}.caveats', ','.join(f.caveats) or '-')


probe('sql', 'ORDERS .> FILTER(_["AMOUNT"] > 10) .> MAP(RECORD("id", _["ID"], "amount", _["AMOUNT"]))')
probe('hybrid', 'ORDERS .> SORT_BY(_["AMOUNT"]) .> FILTER(_K > 1)')
probe('memory', 'A += 1; ORDERS .> TAKE(1)')
# A three-argument LINK in the continuation: the source variable is the
# relation's own name, and a rebound source still is (sqlt plan cases).
probe('link', 'ORDERS .> SORT_BY(_["AMOUNT"]) .> LINK(CUSTOMERS, _1["CUSTOMER_ID"] == _2["ID"])')
probe('rebind', 'ORDERS = ORDERS .> DROP(2); ORDERS .> SORT_BY(_["AMOUNT"]) .> LINK(CUSTOMERS, _1["CUSTOMER_ID"] == _2["ID"])')
# The canonical flag is public: an application (and the SQL oracle) reads it
# to know the fragment promised a spelling, not only a value (SEL-0058).
fragment_probe('canon.postgresql', 'postgresql', 'CANON(1.50)')
fragment_probe('canon.mariadb', 'mariadb', 'CANON(1.50)')
fragment_probe('canon.sqlite', 'sqlite', 'CANON(1.50)')
fragment_probe('abs.postgresql', 'postgresql', 'ABS(1.50)')

# --- host functions with a SQL spelling (spec §8.1, sql/MAP.md §4.7) ----------
# The registration order, the caveat, strict mode, a LIST argument, a builder,
# reset() and a re-registration, through this host's own spelling of the API.
from sel.sql import Fragment, SqlError, map as sqlmap                 # noqa: E402

HOST = {'T': Binding.column('title', 't', 'TEXT')}


def attempt(fn):
    try:
        fn()
        return 'accepted'
    except SqlError as e:
        return f'SqlError {e.code}'
    except RuntimeError:
        return 'refused'


say('host.spell.before-register', attempt(lambda: sqlmap.define(
    'postgresql', 'funcs', 'HSLUG', {'tpl': 'slug({0})', 'ret': 'TEXT'})))
sel.register_function('HSLUG', 1, 1, lambda a: sel.Value.text('local:' + a.text(0)))
sqlmap.define('postgresql', 'funcs', 'HSLUG', {'tpl': 'slug({0})', 'ret': 'TEXT', 'args': ['TEXT']})
spelled = Sql.translate(sel.compile('HSLUG(T) $== "x"'), 'postgresql', HOST)
say('host.spell.condition', spelled.as_condition())
say('host.spell.caveats', ','.join(spelled.caveats) or '-')
say('host.spell.strict', attempt(lambda: Sql.translate(
    sel.compile('HSLUG(T)'), 'postgresql', HOST, {'strict': True})))
say('host.spell.other-dialect', attempt(lambda: Sql.translate(sel.compile('HSLUG(T)'), 'mariadb', HOST)))

sel.register_function('HHAS', 2, 2, lambda a: sel.Value.bool(False))
sqlmap.define('postgresql', 'funcs', 'HHAS',
              {'tpl': '({1} = ANY(ARRAY[{0}]))', 'ret': 'BOOL', 'args': ['LIST', 'TEXT']})
listed = Sql.translate(sel.compile('HHAS(("a", "b"), "c")'), 'postgresql', HOST)
say('host.spell.list.params', listed.as_condition('params'))
say('host.spell.list.bound', ','.join(v.dump() for v in listed.bindings()))

sel.register_function('HWRAP', 1, 1, lambda a: a.val(0).clone())
sqlmap.define_builder('postgresql', 'funcs', 'HWRAP',
                      lambda emit, args, _at: Fragment(['wrap(', *args[0].parts, ')'], 'TEXT', emit.dialect()))
say('host.spell.builder', Sql.translate(sel.compile('HWRAP(T)'), 'postgresql', HOST).as_value())

sel.register_function('HSLUG', 1, 2, lambda a: sel.Value.text('local:' + a.text(0)))
say('host.spell.reregistered-arity', attempt(lambda: Sql.translate(sel.compile('HSLUG(T)'), 'postgresql', HOST)))
sel.register_function('HSLUG', 1, 1, lambda a: sel.Value.text('local:' + a.text(0)))
say('host.spell.arity-restored', attempt(lambda: Sql.translate(sel.compile('HSLUG(T)'), 'postgresql', HOST)))
sqlmap.reset()
say('host.spell.after-reset', attempt(lambda: Sql.translate(sel.compile('HSLUG(T)'), 'postgresql', HOST)))
say('host.spell.after-reset.local', sel.evaluate('HSLUG("A")').as_text())

# --- rendering and registration state ------------------------------------
# The questions a snapshot of ONE translation cannot ask: what an unknown render
# mode does when there is nothing to bind, and whether a refused dialect stays
# refused. `refuses` collapses the host's own error classes, which differ, to the
# one thing the contract says: the call did not produce SQL.
def refuses(fn):
    try:
        fn()
        return 'accepted'
    except Exception:                                               # noqa: BLE001
        return 'refused'


ONE = {'C': Binding.column('c', 't', 'NUM')}
zero = Sql.translate(sel.compile('C > C'), 'mariadb', ONE)
onelit = Sql.translate(sel.compile('C > 1'), 'mariadb', ONE)
say('render.mode.valid.zero-slots', refuses(lambda: zero.as_value('params')))
say('render.mode.bogus.zero-slots', refuses(lambda: zero.as_value('bogus')))
say('render.mode.bogus.zero-slots.condition', refuses(lambda: zero.as_condition('bogus')))
say('render.mode.bogus.with-slot', refuses(lambda: Sql.translate(sel.compile('C > "x"'), 'mariadb', ONE).as_value('bogus')))
say('render.mode.bogus.literal-number', refuses(lambda: onelit.as_value('bogus')))
sqlmap.define_dialect('probe-badguard', {'extends': 'postgresql', 'version': '16', 'lexical': {
    'numericGuard': "CASE WHEN ({textCast:0} ~ '^.*$') THEN CAST({0} AS NUMERIC) ELSE NULL END"}})
NAMED = {'N': Binding.column('n', 't', 'TEXT')}
for k in (1, 2, 3):
    say(f'guard.reuse.{k}', refuses(lambda: Sql.translate(sel.compile('N + 1'), 'probe-badguard', NAMED)))
sqlmap.reset()
say('guard.reuse.after-reset', refuses(lambda: Sql.translate(sel.compile('N + 1'), 'probe-badguard', NAMED)))

# A builder registered on `skel` has no template to fill: the LIMIT it would
# spell is refused, so the TAKE stays in memory (sql/MAP.md §5). Python raised a
# KeyError out of the planner.
sqlmap.define_builder('mariadb', 'skel', 'limit',
                      lambda emit, args, _at: Fragment(['LIMIT 1'], 'TEXT', emit.dialect()))
plan = Sql.plan_hybrid(sel.compile('ORDERS .> TAKE(2)'), 'mariadb', BINDINGS)
say('skel.builder.plan', 'pure_sql' if plan.pure_sql else 'pure_memory' if plan.pure_memory else 'hybrid')
say('skel.builder.statement', plan.sql_statement.as_statement() if plan.sql_statement is not None else '-')
sqlmap.reset()

sys.stdout.reconfigure(encoding='utf-8', newline='\n')
sys.stdout.write('\n'.join(out) + '\n')
