"""Local planning exercise only: SQLite and OData strings, never Dataverse calls."""
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from tempfile import TemporaryDirectory
from urllib.parse import urlencode, parse_qs, urlsplit
import re
import sqlite3
import uuid
from threading import Barrier

if sqlite3.sqlite_version_info < (3, 24, 0):
    raise RuntimeError("SQLite 3.24 or newer required")

checks = 0


def check(condition, label):
    global checks
    if not condition:
        raise AssertionError(label)
    checks += 1
    print('PASS', label)


def stable_key(database, scope, source_id, gate=None):
    if not scope or source_id is None or not str(source_id).strip():
        raise ValueError('incomplete source identity')
    with sqlite3.connect(database, timeout=10) as connection:
        if gate is not None:
            gate.wait(timeout=10)
        connection.execute(
            'INSERT INTO correspondence VALUES (?, ?, ?) ON CONFLICT(scope, source_id) DO NOTHING',
            (scope, str(source_id), uuid.uuid4().hex),
        )
        return connection.execute(
            'SELECT external_key FROM correspondence WHERE scope=? AND source_id=?',
            (scope, str(source_id)),
        ).fetchone()[0]


def upsert_plan(key, raw_ref, status):
    if status != 'Active':
        raise ValueError('alternate key index unavailable')
    if not re.fullmatch('[0-9a-f]{32}', key):
        raise ValueError('key outside this exercise contract')
    return {
        'method': 'PATCH',
        'path': "/api/data/v9.2/accounts(new_external_key='" + key + "')",
        'body': {'new_raw_ref': raw_ref, 'name': 'Fixture client'},
    }


def filter_plan(scope, raw_ref):
    literal = lambda value: "'" + value.replace("'", "''") + "'"
    expression = 'new_source eq ' + literal(scope) + ' and new_raw_ref eq ' + literal(raw_ref)
    return '/api/data/v9.2/accounts?' + urlencode({
        '$select': 'accountid,new_source,new_raw_ref', '$filter': expression, '$top': 2,
    })


def choose_existing(rows, scope, raw_ref):
    # Candidate rows are supplied fixtures, not results from a Dataverse request.
    if len(rows) != 1:
        raise ValueError('zero or ambiguous candidates: no create decision')
    row = rows[0]
    if row['new_source'] != scope or row['new_raw_ref'] != raw_ref:
        raise ValueError('candidate differs from exact source identity')
    uuid.UUID(row['accountid'])
    return {
        'method': 'PATCH',
        'path': '/api/data/v9.2/accounts(' + row['accountid'] + ')',
        'headers': {'If-Match': '*'},
    }


def refuses(action, label):
    try:
        action()
    except ValueError:
        check(True, label)
    else:
        raise AssertionError(label)


with TemporaryDirectory(prefix='dataverse-key-local-') as directory:
    database = Path(directory) / 'mapping.sqlite'
    with sqlite3.connect(database) as connection:
        connection.execute('''CREATE TABLE correspondence (
            scope TEXT NOT NULL, source_id TEXT NOT NULL,
            external_key TEXT NOT NULL UNIQUE,
            PRIMARY KEY(scope, source_id)
        )''')
    # Independent connections: competing assignments must return the stored winner.
    gate = Barrier(2)
    with ThreadPoolExecutor(max_workers=2) as workers:
        keys = list(workers.map(lambda _: stable_key(database, 'ERP-A', '901', gate), range(2)))
    first = keys[0]
    check(keys[0] == keys[1], 'two assignments return one stored key')
    check(re.fullmatch('[0-9a-f]{32}', first) is not None, 'key follows local safe alphabet')
    check(stable_key(database, 'ERP-A', '901') == first, 'reopened connection preserves key')
    other_source = stable_key(database, 'ERP-B', '901')
    check(other_source != first, 'same source id in another company remains distinct')
    second = stable_key(database, 'ERP-A', '902')
    check(second != first, 'different ERP objects keep separate identities')
    naive = lambda value: re.sub('[^A-Za-z0-9]', '', value)
    check(naive('CLI/42') == naive('CLI42'), 'lossy normalization creates fixture collision')
    p1 = upsert_plan(first, 'CLI/42', 'Active')
    p2 = upsert_plan(second, 'CLI42', 'Active')
    check(p1['path'] != p2['path'], 'colliding display references yield distinct targets')
    check(p1['body']['new_raw_ref'] == 'CLI/42', 'display reference preserved')
    check('new_external_key' not in p1['body'], 'alternate key omitted from body')
    check(upsert_plan(first, 'CLI/42-renamed', 'Active')['path'] == p1['path'], 'renaming reference preserves object identity')
    for status in ['Pending', 'In Progress', 'Failed']:
        refuses(lambda status=status: upsert_plan(first, 'CLI/42', status), 'block index ' + status)
    for source_id in [None, '']:
        refuses(lambda source_id=source_id: stable_key(database, 'ERP-A', source_id), 'reject missing source id ' + repr(source_id))
    refuses(lambda: stable_key(database, '', '901'), 'reject missing company scope')
    refuses(lambda: upsert_plan('CLI/42', 'CLI/42', 'Active'), 'raw reference not admitted as safe key')
    raw = "CLI/O'Neil&50%"
    path = filter_plan('ERP-A', raw)
    query = parse_qs(urlsplit(path).query)
    check(query['$filter'] == ["new_source eq 'ERP-A' and new_raw_ref eq 'CLI/O''Neil&50%'"], 'OData escaping and query encoding preserve raw reference')
    check(query['$top'] == ['2'], 'candidate limit does not silently choose first result')
    guid = '00000000-0000-4000-8000-000000000901'
    row = {'accountid': guid, 'new_source': 'ERP-A', 'new_raw_ref': raw}
    resolved = choose_existing([row], 'ERP-A', raw)
    check(guid in resolved['path'], 'unique exact fixture addressed by GUID')
    check(resolved['headers'] == {'If-Match': '*'}, 'existing-target plan prohibits implicit create')
    refuses(lambda: choose_existing([], 'ERP-A', raw), 'zero candidates does not authorize creation')
    refuses(lambda: choose_existing([row, row], 'ERP-A', raw), 'two candidates block selection')
    refuses(lambda: choose_existing([row], 'ERP-B', raw), 'wrong company blocks selection')
    refuses(lambda: choose_existing([row], 'ERP-A', raw.lower()), 'case-different fixture blocks exact selection')
    with sqlite3.connect(database) as connection:
        check(connection.execute('SELECT count(*) FROM correspondence').fetchone()[0] == 3, 'only three identities stored')
print('CHECKS', checks)
print('Dataverse requests: 0; native writes: 0; no token required.')