package gudusoft.gsqlparser.lineage.v2.identity;

import gudusoft.gsqlparser.lineage.v2.SqllrException;
import gudusoft.gsqlparser.lineage.v2.json.CanonicalJson;

import java.io.ByteArrayOutputStream;
import java.math.BigInteger;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

/**
 * The identity algorithm family {@code PUBLIC-LINEAGE-ID-V1} ({@code lineage-spec/algorithms/identity-v1.md}).
 *
 * <p>Each algorithm takes its inputs as a JSON value in the shape of the specification's vectors and of the
 * reference implementation ({@code sqllr_ids.py}, {@code compute}), validates them in the reference's order,
 * and returns an {@link IdResult}: the identity status, and the framed input, digest, id and numeric key when
 * an id is computed. Canonical segment keys are inputs; minting them belongs to the producer.</p>
 *
 * @since 4.2.17
 */
public final class IdentityV1 {

    public static final String FAMILY = "PUBLIC-LINEAGE-ID-V1";
    public static final String PATH_PROFILE_V1 = "sqllr-path-1";

    public static final String SOURCE_UNIT = "SourceUnitIdV1";
    public static final String ENDPOINT = "EndpointIdV1";
    public static final String FACT = "FactIdV1";
    public static final String OCCURRENCE_KEY = "OccurrenceKeyV1";
    public static final String ANALYSIS_CONTEXT = "AnalysisContextV1";
    public static final String OCCURRENCE_RECORD = "OccurrenceRecordIdV1";
    public static final String EXPLAIN_NODE_KEY = "ExplainNodeKeyV1";
    public static final String EXPLAIN_NODE_RECORD = "ExplainNodeRecordIdV1";
    public static final String DEPENDENCY_SNAPSHOT = "DependencySnapshotIdV1";
    public static final String METADATA_DIGEST = "MetadataDigestV1";

    /** Algorithm name to the prefix of its ids. */
    public static final Map<String, String> PREFIXES;

    static {
        Map<String, String> prefixes = new LinkedHashMap<String, String>();
        prefixes.put(SOURCE_UNIT, "su1");
        prefixes.put(ENDPOINT, "ep1");
        prefixes.put(FACT, "fact1");
        prefixes.put(OCCURRENCE_KEY, "occk1");
        prefixes.put(ANALYSIS_CONTEXT, "actx1");
        prefixes.put(OCCURRENCE_RECORD, "occr1");
        prefixes.put(EXPLAIN_NODE_KEY, "xnk1");
        prefixes.put(EXPLAIN_NODE_RECORD, "xnr1");
        prefixes.put(DEPENDENCY_SNAPSHOT, "snap1");
        PREFIXES = Collections.unmodifiableMap(prefixes);
    }

    public static final String PORTABLE = "PORTABLE";
    public static final String LOCAL = "LOCAL";
    public static final String UNAVAILABLE_COLLATION = "UNAVAILABLE_COLLATION";
    public static final String UNAVAILABLE_MIXED_PROFILE = "UNAVAILABLE_MIXED_PROFILE";
    public static final String OPAQUE_PRODUCER_LOCAL = "OPAQUE_PRODUCER_LOCAL";

    private static final Pattern ID = Pattern.compile("(su1|ep1|fact1|occk1|actx1|occr1|xnk1|xnr1|snap1)-([0-9a-f]{64})");
    private static final Pattern HEX64 = Pattern.compile("[0-9a-f]{64}");
    /** One step of an {@code sqllr-path-1} path. A whole path is checked step by step: Java's regex engine
     * recurses once per repetition of a group, so one pattern over a long path would exhaust the stack. */
    private static final Pattern PATH_STEP = Pattern.compile("[A-Z][A-Z0-9_]*\\[(0|[1-9][0-9]*)\\]");
    private static final long MAX_ORDINAL = 9007199254740991L;

    private static List<String> list(String... values) {
        return Collections.unmodifiableList(Arrays.asList(values));
    }

    public static final List<String> FACT_KINDS = list("VALUE", "ROW", "DATASET_DEPENDENCY", "CALL");
    public static final List<String> ROW_ROLES = list("FILTER", "JOIN_CONDITION", "GROUP_BY", "HAVING", "ORDER_BY",
            "WINDOW_PARTITION", "WINDOW_ORDER", "DISTINCT", "QUALIFY", "LIMIT", "CONDITIONAL", "ROW_PROPAGATION", "UNKNOWN");
    public static final List<String> NON_PORTABLE = list(LOCAL, UNAVAILABLE_COLLATION, UNAVAILABLE_MIXED_PROFILE,
            OPAQUE_PRODUCER_LOCAL);
    public static final List<String> DIGEST_DOMAINS = list("RAW_BYTES_SHA256", "DECODED_TEXT_SHA256",
            "REDACTED_TEXT_SHA256", "KEYED_TOKEN");
    public static final List<String> METADATA_STATUSES = list("NONE", "PARTIAL", "COMPLETE");
    public static final List<String> ANALYSIS_FEATURES = list("VALUE", "ROW", "DATASET_DEPENDENCY", "CALL",
            "CONSTANT_INPUTS", "DYNAMIC_SQL", "ROUTINE_BODIES", "TEMPORARY_OBJECTS", "WILDCARD_EXPANSION", "NESTED_FIELDS");
    public static final List<String> CAPABILITY_STATUSES = list("SUPPORTED", "UNSUPPORTED", "PARTIAL");
    public static final List<String> BOM_HANDLING = list("NO_BOM", "BOM_STRIPPED");
    public static final List<String> LINE_ENDING_HANDLING = list("PRESERVED", "NORMALIZED_LF");
    public static final List<String> ROUTINE_NAMESPACES = list("CATALOG");
    public static final List<String> ROUTINE_KINDS = list("FUNCTION", "PROCEDURE");
    public static final List<String> SIGNATURE_FORMS = list("NOT_OVERLOADABLE", "PARAMETERS", "CATALOG_OVERLOAD_ID");
    public static final List<String> PARAMETER_MODES = list("IN", "OUT", "INOUT", "VARIADIC");
    public static final List<String> PATH_SEGMENT_KINDS = list("FIELD", "ARRAY_ELEMENT", "MAP_KEY", "MAP_VALUE", "INDEX");
    public static final List<String> CANDIDATE_SET_MEMBERSHIP = list("COMPLETE", "INCOMPLETE");
    public static final List<String> PATH_STEP_CODES = list(
            "SELECT", "INSERT", "UPDATE", "DELETE", "MERGE", "CREATE_TABLE", "CREATE_VIEW", "COPY", "CALL", "EXECUTE",
            "TARGET", "COLUMN_LIST", "SELECT_LIST", "FROM", "JOIN", "ON", "USING", "WHERE", "GROUP_BY", "HAVING",
            "QUALIFY", "ORDER_BY", "LIMIT", "WINDOW", "PARTITION_BY", "WITH", "CTE", "SET_OP", "BRANCH", "SET",
            "VALUES", "WHEN", "RETURNING", "SUBQUERY", "FUNCTION_ARG", "CASE_WHEN", "CASE_THEN", "CASE_ELSE", "PIVOT",
            "LATERAL", "SOURCE", "CREATE_ROUTINE", "RETURN", "BLOCK", "ASSIGNMENT", "ASSIGN", "CONDITION", "CURSOR",
            "FETCH");
    public static final List<String> EXPLAIN_NODE_KINDS = list("RESULT_SET", "CTE", "SUBQUERY", "SET_OPERATION",
            "ROW_SET", "FIELD", "FUNCTION", "CASE", "EXPRESSION", "TERMINAL", "TABLE_FUNCTION", "PIVOT");

    /**
     * Endpoint kind (registry group {@code endpointKinds}) to identity class (Section 5.2.1); a kind mapped to
     * null has no identity class in V1 and is {@code LOCAL}.
     */
    public static final Map<String, String> IDENTITY_CLASS;

    static {
        Map<String, String> classes = new LinkedHashMap<String, String>();
        for (String kind : new String[] {"RELATION", "TABLE", "VIEW", "MATERIALIZED_VIEW", "EXTERNAL_TABLE", "SYNONYM"}) {
            classes.put(kind, "RELATION");
        }
        classes.put("FIELD", "FIELD");
        classes.put("ROW_SET", "ROW_SET");
        classes.put("ROUTINE", "ROUTINE");
        for (String kind : new String[] {"QUERY_RESULT", "TEMP_TABLE", "TABLE_VARIABLE", "CTE", "VARIABLE", "PARAMETER",
                "SEQUENCE", "STAGE", "STORAGE_LOCATION"}) {
            classes.put(kind, null);
        }
        IDENTITY_CLASS = Collections.unmodifiableMap(classes);
    }

    /** Precedence of Section 4.2: the higher rank wins. */
    private static final Map<String, Integer> STATUS_RANK = new HashMap<String, Integer>();
    private static final Map<String, String> KEY_UNAVAILABLE = new HashMap<String, String>();

    static {
        STATUS_RANK.put(PORTABLE, 0);
        STATUS_RANK.put(UNAVAILABLE_COLLATION, 1);
        STATUS_RANK.put(UNAVAILABLE_MIXED_PROFILE, 2);
        STATUS_RANK.put(LOCAL, 3);
        STATUS_RANK.put(OPAQUE_PRODUCER_LOCAL, 4);
        KEY_UNAVAILABLE.put("COLLATION", UNAVAILABLE_COLLATION);
        KEY_UNAVAILABLE.put("MIXED_PROFILE", UNAVAILABLE_MIXED_PROFILE);
    }

    private IdentityV1() {
    }

    // ---------------------------------------------------------------------------------------------
    // Results
    // ---------------------------------------------------------------------------------------------

    /** The outcome of one algorithm run. */
    public static final class IdResult {
        private final String status;
        private final byte[] framed;
        private final String sha256Hex;
        private final String id;
        private final String numericKey;

        IdResult(String status, byte[] framed, String sha256Hex, String id, String numericKey) {
            this.status = status;
            this.framed = framed;
            this.sha256Hex = sha256Hex;
            this.id = id;
            this.numericKey = numericKey;
        }

        /** The identity status ({@code PORTABLE}, {@code LOCAL}, ...). */
        public String getStatus() {
            return status;
        }

        /** The framed input (the identity content), or null when no id is computed. */
        public byte[] getFramed() {
            return framed == null ? null : framed.clone();
        }

        public String getFramedHex() {
            return framed == null ? null : Framing.hex(framed);
        }

        public String getSha256Hex() {
            return sha256Hex;
        }

        /** The id ({@code ep1-…}), the bare digest for {@code MetadataDigestV1}, or null. */
        public String getId() {
            return id;
        }

        /** {@code NumericKeyV1} of the id, or null. */
        public String getNumericKey() {
            return numericKey;
        }
    }

    /** The status and the fields an algorithm frames; null fields when no id is computed. */
    private static final class Outcome {
        final String status;
        final List<Object> fields;

        Outcome(String status, List<Object> fields) {
            this.status = status;
            this.fields = fields;
        }
    }

    private static Outcome portable(Object... fields) {
        return new Outcome(PORTABLE, Arrays.asList(fields));
    }

    private static Outcome noId(String status) {
        return new Outcome(status, null);
    }

    /**
     * Runs one algorithm on its inputs.
     *
     * @param algorithm one of the names of Section 2, such as {@code EndpointIdV1}
     * @throws SqllrException with an {@code ID_*} code (or a {@code SCHEMA_*} code for JSON that canonical JSON
     *                        cannot carry) when the inputs are rejected
     */
    public static IdResult compute(String algorithm, Object inputs) {
        Outcome outcome;
        if (SOURCE_UNIT.equals(algorithm)) {
            outcome = sourceUnitId(inputs);
        } else if (ENDPOINT.equals(algorithm)) {
            outcome = endpointId(inputs);
        } else if (FACT.equals(algorithm)) {
            outcome = factId(inputs);
        } else if (OCCURRENCE_KEY.equals(algorithm)) {
            outcome = occurrenceKey(inputs);
        } else if (ANALYSIS_CONTEXT.equals(algorithm)) {
            outcome = analysisContext(inputs);
        } else if (OCCURRENCE_RECORD.equals(algorithm)) {
            Map<?, ?> in = members(inputs, list("occurrenceKey", "analysisContext"), NONE, "OccurrenceRecordIdV1 input");
            outcome = portable(ref(in.get("occurrenceKey"), "occk1", "occurrenceKey"),
                    ref(in.get("analysisContext"), "actx1", "analysisContext"));
        } else if (EXPLAIN_NODE_KEY.equals(algorithm)) {
            outcome = explainNodeKey(inputs);
        } else if (EXPLAIN_NODE_RECORD.equals(algorithm)) {
            Map<?, ?> in = members(inputs, list("nodeKey", "analysisContext"), NONE, "ExplainNodeRecordIdV1 input");
            outcome = portable(ref(in.get("nodeKey"), "xnk1", "nodeKey"),
                    ref(in.get("analysisContext"), "actx1", "analysisContext"));
        } else if (DEPENDENCY_SNAPSHOT.equals(algorithm)) {
            Map<?, ?> in = members(inputs, list("manifest"), NONE, "DependencySnapshotIdV1 input");
            if (!(in.get("manifest") instanceof Map<?, ?>)) {
                throw invalid("manifest must be an object");
            }
            outcome = portable(CanonicalJson.jcsText(in.get("manifest")));
        } else if (METADATA_DIGEST.equals(algorithm)) {
            outcome = metadataDigest(inputs);
        } else {
            throw invalid("unknown algorithm " + algorithm);
        }
        if (outcome.fields == null) {
            return new IdResult(outcome.status, null, null, null, null);
        }
        return mint(algorithm, outcome.status, outcome.fields);
    }

    private static IdResult mint(String algorithm, String status, List<Object> fields) {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        Framing.frameInto("SQL-LR/" + algorithm, out);
        for (Object field : fields) {
            Framing.frameInto(field, out);
        }
        byte[] framed = out.toByteArray();
        String digest = Framing.hex(sha256(framed));
        if (METADATA_DIGEST.equals(algorithm)) {
            return new IdResult(status, framed, digest, digest, null);
        }
        return new IdResult(status, framed, digest, PREFIXES.get(algorithm) + "-" + digest, digest.substring(0, 32));
    }

    static byte[] sha256(byte[] data) {
        try {
            return MessageDigest.getInstance("SHA-256").digest(data);
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException(e);
        }
    }

    /**
     * {@code NumericKeyV1} (Section 5.10): the first 16 bytes of the digest of an id, as 32 lowercase
     * hexadecimal characters.
     *
     * @throws SqllrException {@code ID_INVALID_REFERENCE} for a string that is not an id of the family
     */
    public static String numericKey(Object id) {
        if (!(id instanceof String) || !ID.matcher((String) id).matches()) {
            throw new SqllrException("ID_INVALID_REFERENCE", "not a PUBLIC-LINEAGE-ID-V1 id: " + id);
        }
        String text = (String) id;
        return text.substring(text.indexOf('-') + 1, text.indexOf('-') + 33);
    }

    /**
     * The importer rule of Section 5.10: two different ids with one numeric key are a fatal collision.
     *
     * @throws SqllrException {@code ID_NUMERIC_KEY_COLLISION}
     */
    public static void checkNumericKeys(Collection<?> ids) {
        Map<String, Object> seen = new HashMap<String, Object>();
        for (Object id : ids) {
            String key = numericKey(id);
            Object earlier = seen.get(key);
            if (earlier != null && !earlier.equals(id)) {
                throw new SqllrException("ID_NUMERIC_KEY_COLLISION", earlier + " and " + id + " share " + key);
            }
            seen.put(key, id);
        }
    }

    /** True for a well-formed id with this prefix. */
    public static boolean isId(Object value, String prefix) {
        return value instanceof String && ID.matcher((String) value).matches() && ((String) value).startsWith(prefix + "-");
    }

    // ---------------------------------------------------------------------------------------------
    // Input checks, in the reference's order
    // ---------------------------------------------------------------------------------------------

    private static final List<String> NONE = Collections.emptyList();

    private static SqllrException invalid(String detail) {
        return new SqllrException("ID_INVALID_INPUT", detail);
    }

    /** The members of an input object: required ones present, no unknown ones ({@code evidence} and {@code note} allowed). */
    private static Map<?, ?> members(Object value, List<String> required, List<String> optional, String what) {
        if (!(value instanceof Map<?, ?>)) {
            throw invalid(what + " must be an object");
        }
        Map<?, ?> obj = (Map<?, ?>) value;
        for (String name : required) {
            if (!obj.containsKey(name)) {
                throw invalid(what + " lacks member " + name);
            }
        }
        for (Object name : obj.keySet()) {
            if (!required.contains(name) && !optional.contains(name) && !"evidence".equals(name) && !"note".equals(name)) {
                throw invalid(what + " has unknown member " + name);
            }
        }
        return obj;
    }

    private static String text(Map<?, ?> obj, String name) {
        return text(obj, name, false, false);
    }

    private static String text(Map<?, ?> obj, String name, boolean allowEmpty, boolean nullable) {
        Object value = obj.get(name);
        if (value == null && nullable) {
            return null;
        }
        if (!(value instanceof String)) {
            throw invalid(name + " must be a string");
        }
        Framing.utf8((String) value);
        if (((String) value).isEmpty() && !allowEmpty) {
            throw new SqllrException("ID_EMPTY_NAME", name + " must not be empty");
        }
        return (String) value;
    }

    /** An integer in 0..2^53-1, or null when {@code value} is not one. */
    private static Long smallInteger(Object value) {
        if (value instanceof Long || value instanceof Integer || value instanceof Short || value instanceof Byte) {
            long v = ((Number) value).longValue();
            return v >= 0 && v <= MAX_ORDINAL ? Long.valueOf(v) : null;
        }
        if (value instanceof BigInteger) {
            BigInteger big = (BigInteger) value;
            return big.signum() >= 0 && big.compareTo(BigInteger.valueOf(MAX_ORDINAL)) <= 0 ? Long.valueOf(big.longValue()) : null;
        }
        return null;
    }

    private static Long ordinal(Map<?, ?> obj, String name) {
        Long value = smallInteger(obj.get(name));
        if (value == null) {
            throw invalid(name + " must be an integer in 0..2^53-1");
        }
        return value;
    }

    private static Code code(Map<?, ?> obj, String name, List<String> allowed) {
        Code value = Code.of(obj.get(name));
        if (allowed != null && !allowed.contains(value.value())) {
            throw new SqllrException("ID_INVALID_CODE", name + ": " + value.value() + " is not one of " + allowed);
        }
        return value;
    }

    /** An id-valued input, framed as the string of its full text (Section 3.6). */
    private static String ref(Object value, String prefix, String what) {
        if (isId(value, prefix)) {
            return (String) value;
        }
        throw new SqllrException("ID_INVALID_REFERENCE", what + " must be a " + prefix + "- id");
    }

    /** A reference given as {@code {"status": <non-portable status>}} instead of an id, or null. */
    private static String statusOf(Object value) {
        if (value instanceof Map<?, ?>) {
            Map<?, ?> obj = members(value, list("status"), NONE, "non-portable reference");
            Object status = obj.get("status");
            if (!(status instanceof String) || !NON_PORTABLE.contains(status)) {
                throw invalid("unknown identity status " + status);
            }
            return (String) status;
        }
        return null;
    }

    private static String worst(List<String> statuses) {
        String out = null;
        for (String status : statuses) {
            if (out == null || STATUS_RANK.get(status) > STATUS_RANK.get(out)) {
                out = status;
            }
        }
        return out == null ? PORTABLE : out;
    }

    private static List<Object> pathScope(Object value) {
        Map<?, ?> obj = members(value, list("profile", "producer"), NONE, "pathScope");
        String profile = text(obj, "profile");
        String producer = text(obj, "producer", false, true);
        if (PATH_PROFILE_V1.equals(profile)) {
            if (producer != null) {
                throw new SqllrException("ID_INVALID_PATH_SCOPE", "producer must be null for " + PATH_PROFILE_V1);
            }
        } else if (producer == null) {
            throw new SqllrException("ID_INVALID_PATH_SCOPE", "a producer path profile needs the producer name");
        }
        return Arrays.<Object>asList(profile, producer);
    }

    private static String semanticPath(List<Object> scope, Object value) {
        if (!(value instanceof String) || ((String) value).isEmpty()) {
            throw new SqllrException("ID_INVALID_PATH", "path must be a non-empty string");
        }
        String path = (String) value;
        Framing.utf8(path);
        if (PATH_PROFILE_V1.equals(scope.get(0))) {
            String[] steps = path.split("/", -1);
            for (String step : steps) {
                if (!PATH_STEP.matcher(step).matches()) {
                    throw new SqllrException("ID_INVALID_PATH", path + " is not a " + PATH_PROFILE_V1 + " path");
                }
            }
            for (String step : steps) {
                if (!PATH_STEP_CODES.contains(step.substring(0, step.indexOf('[')))) {
                    throw new SqllrException("ID_INVALID_PATH", step + " is not a step code of " + PATH_PROFILE_V1);
                }
            }
        }
        return path;
    }

    private static List<Object> contentDigest(Object value) {
        Map<?, ?> obj = members(value, list("domain", "value"), list("tokenDomain"), "content digest");
        Code domain = code(obj, "domain", DIGEST_DOMAINS);
        if ("KEYED_TOKEN".equals(domain.value())) {
            return Arrays.<Object>asList(domain, text(obj, "tokenDomain"), text(obj, "value"));
        }
        if (obj.get("tokenDomain") != null) {
            throw new SqllrException("ID_INVALID_DIGEST", "tokenDomain is allowed only for KEYED_TOKEN");
        }
        Object digest = obj.get("value");
        if (!(digest instanceof String) || !HEX64.matcher((String) digest).matches()) {
            throw new SqllrException("ID_INVALID_DIGEST", "a SHA-256 digest is 64 lowercase hex characters");
        }
        return Arrays.<Object>asList(domain, null, digest);
    }

    private static String hex64(Map<?, ?> obj, String name, boolean nullable) {
        Object value = obj.get(name);
        if (value == null && nullable) {
            return null;
        }
        if (!(value instanceof String) || !HEX64.matcher((String) value).matches()) {
            throw new SqllrException("ID_INVALID_DIGEST", name + " must be 64 lowercase hex characters");
        }
        return (String) value;
    }

    /** Framed name segments and the statuses that prevent an id (Section 3.9). */
    private static final class Segments {
        final List<Object> framed = new ArrayList<Object>();
        final List<String> blocked = new ArrayList<String>();
    }

    private static Segments nameSegments(Object value) {
        if (!(value instanceof List<?>) || ((List<?>) value).size() < 2) {
            throw invalid("a full name has an AUTHORITY segment and an object segment");
        }
        List<?> items = (List<?>) value;
        Segments out = new Segments();
        for (int position = 0; position < items.size(); position++) {
            Object item = items.get(position);
            Object kind = item instanceof Map<?, ?> ? ((Map<?, ?>) item).get("kind") : null;
            if (!"NAMED".equals(kind) && !"PRODUCER_DEFAULT".equals(kind)) {
                throw invalid("segment.kind must be NAMED or PRODUCER_DEFAULT");
            }
            Map<?, ?> seg = (Map<?, ?>) item;
            Code role = code(seg, "role", null);
            if ("AUTHORITY".equals(role.value()) != (position == 0)) {
                throw invalid("AUTHORITY is the first segment and only the first");
            }
            if ("PRODUCER_DEFAULT".equals(kind)) {
                if (seg.containsKey("canonicalKey") || seg.containsKey("keyUnavailable")) {
                    throw new SqllrException("ID_INVALID_DEFAULT_SEGMENT", "a PRODUCER_DEFAULT segment has no key");
                }
                if (position == items.size() - 1) {
                    throw invalid("the object segment is never defaulted");
                }
                members(seg, list("role", "kind"), NONE, "segment");
                out.framed.add(Arrays.<Object>asList(role, Code.of("PRODUCER_DEFAULT"), null));
                continue;
            }
            if (seg.containsKey("keyUnavailable")) {
                members(seg, list("role", "kind", "keyUnavailable"), NONE, "segment");
                Object reason = seg.get("keyUnavailable");
                if (!KEY_UNAVAILABLE.containsKey(reason) || position == 0) {
                    throw invalid("keyUnavailable must be COLLATION or MIXED_PROFILE");
                }
                out.blocked.add(KEY_UNAVAILABLE.get(reason));
                continue;
            }
            members(seg, list("role", "kind", "canonicalKey"), NONE, "segment");
            // For AUTHORITY the key is the caller-declared name, byte for byte (Section 3.9).
            out.framed.add(Arrays.<Object>asList(role, Code.of("NAMED"), text(seg, "canonicalKey")));
        }
        return out;
    }

    /** A field path (outermost first, starting with {@code FIELD}) and the statuses that prevent an id. */
    private static Segments fieldPath(Object value) {
        if (!(value instanceof List<?>) || ((List<?>) value).isEmpty()) {
            throw new SqllrException("ID_INVALID_FIELD_PATH", "a field path has at least one segment");
        }
        List<?> items = (List<?>) value;
        Segments out = new Segments();
        for (int position = 0; position < items.size(); position++) {
            Object item = items.get(position);
            Object kindValue = item instanceof Map<?, ?> ? ((Map<?, ?>) item).get("kind") : null;
            if (!(kindValue instanceof String) || !PATH_SEGMENT_KINDS.contains(kindValue)) {
                throw new SqllrException("ID_INVALID_FIELD_PATH", "unknown path segment kind");
            }
            Map<?, ?> seg = (Map<?, ?>) item;
            Code kind = Code.of(kindValue);
            if (position == 0 && !"FIELD".equals(kind.value())) {
                throw new SqllrException("ID_INVALID_FIELD_PATH", "the first path segment must be FIELD");
            }
            if ("FIELD".equals(kind.value())) {
                if (seg.containsKey("keyUnavailable")) {
                    members(seg, list("kind", "keyUnavailable"), NONE, "path segment");
                    Object reason = seg.get("keyUnavailable");
                    if (!KEY_UNAVAILABLE.containsKey(reason)) {
                        throw invalid("keyUnavailable must be COLLATION or MIXED_PROFILE");
                    }
                    out.blocked.add(KEY_UNAVAILABLE.get(reason));
                    continue;
                }
                members(seg, list("kind", "canonicalKey"), NONE, "path segment");
                // A column key is never empty; a nested key of semi-structured data may be.
                out.framed.add(Arrays.<Object>asList(kind, text(seg, "canonicalKey", position > 0, false)));
            } else if ("INDEX".equals(kind.value())) {
                members(seg, list("kind", "index"), NONE, "path segment");
                Long index = smallInteger(seg.get("index"));
                if (index == null) {
                    throw new SqllrException("ID_INVALID_FIELD_PATH", "INDEX is a 0-based integer");
                }
                out.framed.add(Arrays.<Object>asList(kind, index));
            } else {
                members(seg, list("kind"), NONE, "path segment");
                out.framed.add(Arrays.<Object>asList(kind, null));
            }
        }
        return out;
    }

    private static List<Object> routineSignature(Object value) {
        Map<?, ?> obj = members(value, list("form"), list("parameters", "overloadId"), "signature");
        Code form = code(obj, "form", SIGNATURE_FORMS);
        if ("NOT_OVERLOADABLE".equals(form.value())) {
            members(obj, list("form"), NONE, "signature");
            return Arrays.<Object>asList(form, null);
        }
        if ("CATALOG_OVERLOAD_ID".equals(form.value())) {
            members(obj, list("form", "overloadId"), NONE, "signature");
            return Arrays.<Object>asList(form, text(obj, "overloadId"));
        }
        members(obj, list("form", "parameters"), NONE, "signature");
        Object params = obj.get("parameters");
        if (!(params instanceof List<?>)) {
            throw new SqllrException("ID_INVALID_SIGNATURE", "parameters must be a list (possibly empty)");
        }
        List<Object> framed = new ArrayList<Object>();
        for (Object param : (List<?>) params) {
            Map<?, ?> p = members(param, list("mode", "canonicalTypeKey"), NONE, "parameter");
            framed.add(Arrays.<Object>asList(code(p, "mode", PARAMETER_MODES), text(p, "canonicalTypeKey")));
        }
        return Arrays.<Object>asList(form, framed);
    }

    // ---------------------------------------------------------------------------------------------
    // The algorithms (Section 5)
    // ---------------------------------------------------------------------------------------------

    private static Outcome sourceUnitId(Object value) {
        Object kind = value instanceof Map<?, ?> ? ((Map<?, ?>) value).get("kind") : null;
        if ("LOCAL".equals(kind)) {
            members(value, list("kind"), NONE, "SourceUnitIdV1 input");
            return noId(LOCAL);
        }
        if ("NAMED".equals(kind)) {
            Map<?, ?> in = members(value, list("kind", "namespace", "logicalName"), NONE, "SourceUnitIdV1 input");
            return portable(Code.of("NAMED"), text(in, "namespace"), text(in, "logicalName"));
        }
        if ("CONTENT_ADDRESSED".equals(kind)) {
            Map<?, ?> in = members(value, list("kind", "contextDomain", "contentDigest"), NONE, "SourceUnitIdV1 input");
            Code code = Code.of("CONTENT_ADDRESSED");
            String domain = text(in, "contextDomain");
            return portable(code, domain, contentDigest(in.get("contentDigest")));
        }
        if ("DERIVED".equals(kind)) {
            Map<?, ?> in = members(value, list("kind", "parentUnit", "statementOrdinal", "pathScope", "sitePath",
                    "siteOrdinal", "variantOrdinal"), NONE, "SourceUnitIdV1 input");
            String parentStatus = statusOf(in.get("parentUnit"));
            if (parentStatus != null) {
                return noId(worst(Arrays.asList(LOCAL, parentStatus)));
            }
            List<Object> scope = pathScope(in.get("pathScope"));
            String parent = ref(in.get("parentUnit"), "su1", "parentUnit");
            Long statement = ordinal(in, "statementOrdinal");
            String sitePath = semanticPath(scope, in.get("sitePath"));
            Long site = ordinal(in, "siteOrdinal");
            Long variant = ordinal(in, "variantOrdinal");
            return portable(Code.of("DERIVED"), parent, statement, scope, sitePath, site, variant);
        }
        throw invalid("SourceUnitIdV1 kind must be NAMED, CONTENT_ADDRESSED, DERIVED or LOCAL");
    }

    private static Outcome endpointId(Object value) {
        if (!(value instanceof Map<?, ?>)) {
            throw invalid("EndpointIdV1 input must be an object");
        }
        Map<?, ?> in = (Map<?, ?>) value;
        Object endpointKind = in.get("endpointKind");
        if (!(endpointKind instanceof String) || !IDENTITY_CLASS.containsKey(endpointKind)) {
            throw new SqllrException("ID_INVALID_CODE", "unknown endpointKind " + endpointKind);
        }
        if (in.containsKey("opaque")) {
            members(in, list("endpointKind", "opaque"), NONE, "EndpointIdV1 input");
            members(in.get("opaque"), list("issuer", "algorithmId", "policyId", "comparisonDomain", "value"), NONE,
                    "opaque identity");
            return noId(OPAQUE_PRODUCER_LOCAL);
        }
        String kind = IDENTITY_CLASS.get(endpointKind);
        if (kind == null) {
            members(in, list("endpointKind"), NONE, "EndpointIdV1 input");
            return noId(LOCAL);
        }
        if (in.containsKey("qualificationSatisfied") && !Boolean.TRUE.equals(in.get("qualificationSatisfied"))) {
            return noId(LOCAL);
        }
        if ("RELATION".equals(kind)) {
            members(in, list("endpointKind", "segments"), list("qualificationSatisfied"), "EndpointIdV1 RELATION input");
            Segments segments = nameSegments(in.get("segments"));
            if (!segments.blocked.isEmpty()) {
                return noId(worst(segments.blocked));
            }
            return portable(Code.of("RELATION"), segments.framed);
        }
        if ("FIELD".equals(kind) || "ROW_SET".equals(kind)) {
            List<String> required = "FIELD".equals(kind) ? list("endpointKind", "owner", "path") : list("endpointKind", "owner");
            members(in, required, list("qualificationSatisfied"), "EndpointIdV1 " + kind + " input");
            List<String> blocked = new ArrayList<String>();
            String ownerStatus = statusOf(in.get("owner"));
            if (ownerStatus != null) {
                blocked.add(ownerStatus);
            }
            if ("ROW_SET".equals(kind)) {
                if (!blocked.isEmpty()) {
                    return noId(worst(blocked));
                }
                return portable(Code.of("ROW_SET"), ref(in.get("owner"), "ep1", "owner"));
            }
            Segments path = fieldPath(in.get("path"));
            blocked.addAll(path.blocked);
            if (!blocked.isEmpty()) {
                return noId(worst(blocked));
            }
            return portable(Code.of("FIELD"), ref(in.get("owner"), "ep1", "owner"), path.framed);
        }
        members(in, list("endpointKind", "routineNamespace", "routineKind", "segments"),
                list("signature", "qualificationSatisfied"), "EndpointIdV1 ROUTINE input");
        Code namespace = code(in, "routineNamespace", ROUTINE_NAMESPACES);
        Code routineKind = code(in, "routineKind", ROUTINE_KINDS);
        Segments segments = nameSegments(in.get("segments"));
        if (in.get("signature") == null) {
            segments.blocked.add(LOCAL); // the overload is not proven (P1-3)
        }
        if (!segments.blocked.isEmpty()) {
            return noId(worst(segments.blocked));
        }
        return portable(Code.of("ROUTINE"), namespace, routineKind, segments.framed, routineSignature(in.get("signature")));
    }

    private static Outcome factId(Object value) {
        Map<?, ?> in = members(value, list("kind", "source", "target"), list("role"), "FactIdV1 input");
        Code kind = code(in, "kind", FACT_KINDS);
        Object role = in.get("role");
        if ("ROW".equals(kind.value())) {
            if (role == null) {
                throw new SqllrException("ID_INVALID_ROLE", "a ROW Fact carries a role");
            }
            role = code(in, "role", ROW_ROLES);
        } else if (role != null) {
            throw new SqllrException("ID_INVALID_ROLE", "only a ROW Fact carries a role");
        }
        String sourceStatus = statusOf(in.get("source"));
        String targetStatus = statusOf(in.get("target"));
        if (sourceStatus != null || targetStatus != null) {
            return noId(LOCAL);
        }
        String source = ref(in.get("source"), "ep1", "source");
        String target = ref(in.get("target"), "ep1", "target");
        return portable(kind, source, target, role);
    }

    /** A side of a FactCandidate subject, or null when a member has no {@code EndpointIdV1}. */
    private static List<Object> candidateSide(Object value) {
        Map<?, ?> obj = members(value, NONE, list("endpoint", "candidates", "membership"), "candidate side");
        if (obj.containsKey("endpoint") == obj.containsKey("candidates")) {
            throw new SqllrException("ID_INVALID_CANDIDATE_SET", "a side has either endpoint or candidates");
        }
        if (obj.containsKey("endpoint")) {
            if (obj.containsKey("membership")) {
                throw new SqllrException("ID_INVALID_CANDIDATE_SET", "membership belongs to a candidate set");
            }
            if (statusOf(obj.get("endpoint")) != null) {
                return null;
            }
            return Arrays.<Object>asList(Code.of("ENDPOINT"), ref(obj.get("endpoint"), "ep1", "endpoint"));
        }
        if (!obj.containsKey("membership")) {
            throw new SqllrException("ID_INVALID_CANDIDATE_SET", "a candidate set declares its membership");
        }
        Code membership = code(obj, "membership", CANDIDATE_SET_MEMBERSHIP);
        Object items = obj.get("candidates");
        if (!(items instanceof List<?>)) {
            throw new SqllrException("ID_INVALID_CANDIDATE_SET", "candidates must be a list");
        }
        List<?> candidates = (List<?>) items;
        if ("COMPLETE".equals(membership.value()) && candidates.size() < 2) {
            throw new SqllrException("ID_INVALID_CANDIDATE_SET", "a COMPLETE candidate set has at least two members");
        }
        for (Object item : candidates) {
            if (statusOf(item) != null) {
                return null;
            }
        }
        List<String> ids = new ArrayList<String>();
        for (Object item : candidates) {
            ids.add(ref(item, "ep1", "candidate"));
        }
        if (new HashSet<String>(ids).size() != ids.size()) {
            throw new SqllrException("ID_INVALID_CANDIDATE_SET", "a candidate set has no duplicates");
        }
        Collections.sort(ids, UTF8_ORDER);
        return Arrays.<Object>asList(Code.of("CANDIDATE_SET"), membership, new ArrayList<Object>(ids));
    }

    /** Unsigned UTF-8 byte order of strings, which is code point order. */
    static final Comparator<String> UTF8_ORDER = new Comparator<String>() {
        @Override
        public int compare(String a, String b) {
            return CanonicalJson.compareUnsigned(Framing.utf8(a), Framing.utf8(b));
        }
    };

    private static List<Object> occurrenceSubject(Object value) {
        Object kind = value instanceof Map<?, ?> ? ((Map<?, ?>) value).get("kind") : null;
        if ("FACT".equals(kind)) {
            Map<?, ?> obj = members(value, list("kind", "fact"), NONE, "subject");
            if (statusOf(obj.get("fact")) != null) {
                return null;
            }
            return Arrays.<Object>asList(Code.of("FACT"), ref(obj.get("fact"), "fact1", "fact"));
        }
        if ("CANDIDATE".equals(kind)) {
            Map<?, ?> obj = members(value, list("kind", "factKind", "source", "target"), list("role"), "subject");
            Code factKind = code(obj, "factKind", FACT_KINDS);
            Object role = obj.get("role");
            if ("ROW".equals(factKind.value())) {
                if (role == null) {
                    throw new SqllrException("ID_INVALID_ROLE", "a ROW candidate carries a role");
                }
                role = code(obj, "role", ROW_ROLES);
            } else if (role != null) {
                throw new SqllrException("ID_INVALID_ROLE", "only a ROW candidate carries a role");
            }
            if (!(obj.get("source") instanceof Map<?, ?>) || !(obj.get("target") instanceof Map<?, ?>)) {
                throw new SqllrException("ID_INVALID_CANDIDATE_SET", "a candidate side must be an object");
            }
            if (!((Map<?, ?>) obj.get("source")).containsKey("candidates")
                    && !((Map<?, ?>) obj.get("target")).containsKey("candidates")) {
                throw new SqllrException("ID_INVALID_CANDIDATE_SET", "a candidate subject has at least one candidate set");
            }
            List<Object> source = candidateSide(obj.get("source"));
            List<Object> target = candidateSide(obj.get("target"));
            if (source == null || target == null) {
                return null;
            }
            return Arrays.<Object>asList(Code.of("CANDIDATE"), factKind, role, source, target);
        }
        throw invalid("subject.kind must be FACT or CANDIDATE");
    }

    private static Outcome occurrenceKey(Object value) {
        Map<?, ?> in = members(value, list("sourceUnit", "statementOrdinal", "subject", "pathScope", "path", "ordinal"),
                NONE, "OccurrenceKeyV1 input");
        List<Object> scope = pathScope(in.get("pathScope"));
        String path = semanticPath(scope, in.get("path"));
        Long statement = ordinal(in, "statementOrdinal");
        Long occurrence = ordinal(in, "ordinal");
        List<Object> subject = occurrenceSubject(in.get("subject"));
        if (statusOf(in.get("sourceUnit")) != null || subject == null) {
            return noId(LOCAL);
        }
        return portable(ref(in.get("sourceUnit"), "su1", "sourceUnit"), statement, subject, scope, path, occurrence);
    }

    private static Outcome metadataDigest(Object value) {
        Map<?, ?> in = members(value, list("artifacts"), NONE, "MetadataDigestV1 input");
        Object artifacts = in.get("artifacts");
        if (!(artifacts instanceof List<?>)) {
            throw invalid("artifacts must be a list");
        }
        List<List<Object>> framed = new ArrayList<List<Object>>();
        for (Object artifact : (List<?>) artifacts) {
            Map<?, ?> a = members(artifact, list("format", "sha256"), NONE, "metadata artifact");
            framed.add(Arrays.<Object>asList(text(a, "format"), hex64(a, "sha256", false)));
        }
        Collections.sort(framed, new Comparator<List<Object>>() {
            @Override
            public int compare(List<Object> x, List<Object> y) {
                int c = UTF8_ORDER.compare((String) x.get(0), (String) y.get(0));
                return c != 0 ? c : UTF8_ORDER.compare((String) x.get(1), (String) y.get(1));
            }
        });
        for (int i = 1; i < framed.size(); i++) {
            if (framed.get(i).equals(framed.get(i - 1))) {
                throw invalid("a metadata artifact is listed twice");
            }
        }
        return portable(new ArrayList<Object>(framed));
    }

    private static List<Object> capabilityList(Object value) {
        if (!(value instanceof Map<?, ?>) || ((Map<?, ?>) value).isEmpty()) {
            throw invalid("capabilities must be a non-empty object");
        }
        Map<?, ?> obj = (Map<?, ?>) value;
        List<String> features = new ArrayList<String>();
        for (Object feature : obj.keySet()) {
            Framing.utf8((String) feature);
            features.add((String) feature);
        }
        Collections.sort(features, UTF8_ORDER);
        List<Object> out = new ArrayList<Object>();
        for (String feature : features) {
            Code name = Code.of(feature);
            if (!ANALYSIS_FEATURES.contains(name.value())) {
                throw new SqllrException("ID_INVALID_CODE", "f: " + feature + " is not one of " + ANALYSIS_FEATURES);
            }
            out.add(Arrays.<Object>asList(name, code(obj, feature, CAPABILITY_STATUSES)));
        }
        return out;
    }

    private static List<Object> decodingList(Object value) {
        if (value == null) {
            return null;
        }
        Map<?, ?> obj = members(value, list("encoding", "bom", "lineEndings"), NONE, "decoding");
        String encoding = text(obj, "encoding");
        Code bom = code(obj, "bom", BOM_HANDLING);
        return Arrays.<Object>asList(encoding, bom, code(obj, "lineEndings", LINE_ENDING_HANDLING));
    }

    private static Outcome analysisContext(Object value) {
        Map<?, ?> in = members(value, list("producerName", "producerVersion", "dependencySnapshot", "dialectId",
                "dialectVersion", "identifierPolicyId", "pathProfile", "effectiveOptionsDigest", "metadataStatus",
                "metadataDigest", "redactionPolicyId", "redactionPolicyVersion", "capabilities", "decoding",
                "sourceContentDigest", "resolutionInputDigests"), list("effectiveOptions", "metadataArtifacts"),
                "AnalysisContextV1 input");
        String optionsDigest = hex64(in, "effectiveOptionsDigest", false);
        if (in.containsKey("effectiveOptions")) {
            String computed = Framing.hex(sha256(CanonicalJson.jcsBytes(in.get("effectiveOptions"))));
            if (!computed.equals(optionsDigest)) {
                throw new SqllrException("ID_MISMATCH", "effectiveOptionsDigest does not match effectiveOptions");
            }
        }
        Code status = code(in, "metadataStatus", METADATA_STATUSES);
        String meta = hex64(in, "metadataDigest", true);
        if ((meta == null) != "NONE".equals(status.value())) {
            throw invalid("metadataDigest is null if and only if metadataStatus is NONE");
        }
        if (in.containsKey("metadataArtifacts")) {
            Map<String, Object> artifacts = new LinkedHashMap<String, Object>();
            artifacts.put("artifacts", in.get("metadataArtifacts"));
            if (!compute(METADATA_DIGEST, artifacts).getId().equals(meta)) {
                throw new SqllrException("ID_MISMATCH", "metadataDigest does not match metadataArtifacts");
            }
        }
        Object inputs = in.get("resolutionInputDigests");
        if (!(inputs instanceof List<?>)) {
            throw invalid("resolutionInputDigests must be a list");
        }
        List<List<Object>> digests = new ArrayList<List<Object>>();
        for (Object digest : (List<?>) inputs) {
            digests.add(contentDigest(digest));
        }
        final Map<List<Object>, byte[]> framedDigests = new HashMap<List<Object>, byte[]>();
        for (List<Object> digest : digests) {
            framedDigests.put(digest, Framing.frame(digest));
        }
        Collections.sort(digests, new Comparator<List<Object>>() {
            @Override
            public int compare(List<Object> x, List<Object> y) {
                return CanonicalJson.compareUnsigned(framedDigests.get(x), framedDigests.get(y));
            }
        });
        for (int i = 1; i < digests.size(); i++) {
            if (Arrays.equals(framedDigests.get(digests.get(i)), framedDigests.get(digests.get(i - 1)))) {
                throw invalid("a resolution input digest is listed twice");
            }
        }
        String producerName = text(in, "producerName");
        String producerVersion = text(in, "producerVersion");
        String snapshot = ref(in.get("dependencySnapshot"), "snap1", "dependencySnapshot");
        String dialectId = text(in, "dialectId");
        String dialectVersion = text(in, "dialectVersion", false, true);
        String policyId = text(in, "identifierPolicyId");
        String pathProfile = text(in, "pathProfile");
        String redactionPolicyId = text(in, "redactionPolicyId");
        String redactionPolicyVersion = text(in, "redactionPolicyVersion");
        List<Object> capabilities = capabilityList(in.get("capabilities"));
        List<Object> decoding = decodingList(in.get("decoding"));
        List<Object> source = contentDigest(in.get("sourceContentDigest"));
        return portable(producerName, producerVersion, snapshot, dialectId, dialectVersion, policyId, pathProfile,
                optionsDigest, status, meta, redactionPolicyId, redactionPolicyVersion, capabilities, decoding, source,
                new ArrayList<Object>(digests));
    }

    private static Outcome explainNodeKey(Object value) {
        Map<?, ?> in = members(value, list("sourceUnit", "statementOrdinal", "pathScope", "path", "nodeKind", "ordinal"),
                NONE, "ExplainNodeKeyV1 input");
        List<Object> scope = pathScope(in.get("pathScope"));
        String path = semanticPath(scope, in.get("path"));
        Long statement = ordinal(in, "statementOrdinal");
        Long occurrence = ordinal(in, "ordinal");
        Code nodeKind = code(in, "nodeKind", EXPLAIN_NODE_KINDS);
        if (statusOf(in.get("sourceUnit")) != null) {
            return noId(LOCAL);
        }
        return portable(ref(in.get("sourceUnit"), "su1", "sourceUnit"), statement, scope, path, nodeKind, occurrence);
    }

    /** The algorithms in the order of Section 2. */
    public static final List<String> ALGORITHMS = list(SOURCE_UNIT, ENDPOINT, FACT, OCCURRENCE_KEY, ANALYSIS_CONTEXT,
            OCCURRENCE_RECORD, EXPLAIN_NODE_KEY, EXPLAIN_NODE_RECORD, DEPENDENCY_SNAPSHOT, METADATA_DIGEST);
}