package gudusoft.gsqlparser.lineage.v2.producer;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import gudusoft.gsqlparser.dlineage.DataFlowAnalyzer;
import gudusoft.gsqlparser.dlineage.dataflow.metadata.MetadataReader;
import gudusoft.gsqlparser.dlineage.dataflow.model.ModelBindingManager;
import gudusoft.gsqlparser.dlineage.dataflow.model.Option;
import gudusoft.gsqlparser.dlineage.dataflow.model.SqlInfo;
import gudusoft.gsqlparser.dlineage.dataflow.model.SqllrObservations;

/**
 * Produces a SQL Lineage Representation (SQL-LR) L1 document from SQL inputs with GSP's lineage engine
 * (plan {@code docs/designs/data-lineage/06-sqllr-m2-producer-plan-zh.md}).
 *
 * <p>The producer runs its own analysis with fixed options ({@link ProducerOptions}) and an observation recorder,
 * builds an ExplainGraph from what the analysis observed, folds it with the shared fold kernel of
 * {@code lineage.v2.fold}, and writes the Facts, Occurrences and writes of the L1 core. It never changes what a
 * legacy {@code DataFlowAnalyzer} run of the caller outputs.</p>
 *
 * <p>The analysis, its projection ({@code lineage.v2.shape}) and {@code dispose()} run on the calling thread; the
 * document is built from the projection after {@code dispose()}, and could be built on any thread (plan 09, slice
 * 1b). The result holds plain values.</p>
 *
 * @since 4.2.17
 */
public final class SqllrProducer {

    private SqllrProducer() {
    }

    /**
     * Analyses the inputs together (DDL and views in one input are visible to the others, in input order) and
     * returns the document.
     *
     * @throws SqllrProducerException when the inputs or the configuration are refused
     */
    public static ProducerResult produce(List<SqllrInput> inputs, ProducerConfig config) {
        return produce(inputs, config, null);
    }

    /** Tests only: {@code prepare} sees the recorder before the analysis runs (failure injection). */
    static ProducerResult produce(List<SqllrInput> inputs, ProducerConfig config,
            java.util.function.Consumer<SqllrObservations> prepare) {
        return produce(inputs, config, prepare, null);
    }

    /**
     * What the last analysis of a production saw (plan 09, Section 5.4), filled by {@link #produce(List,
     * ProducerConfig, java.util.function.Consumer, Capture)} for the correctness runner and the coverage probe. It
     * holds plain values only, so it stays usable after {@code dispose()} and on another thread.
     */
    static final class Capture {
        private final List<String> inputTexts = new ArrayList<String>();
        private final List<String> analysedTexts = new ArrayList<String>();
        private final List<int[]> blanked = new ArrayList<int[]>();
        private int analyses;
        private String legacyXml;
        private String legacyJson;
        private final java.util.Map<Long, Long> relationOwners = new java.util.TreeMap<Long, Long>();
        private final java.util.Map<Long, String> unitTexts = new java.util.LinkedHashMap<Long, String>();
        private ProducerResult result;
        private gudusoft.gsqlparser.dlineage.dataflow.model.SqllrSnapshot snapshot;
        private String snapshotFailure;
        private final java.util.Map<Long, String> ledgerRelations = new java.util.TreeMap<Long, String>();
        private final java.util.Map<Long, String> ledgerElementEvents = new java.util.TreeMap<Long, String>();
        private final java.util.Map<Long, String> ledgerReferenceEvents = new java.util.TreeMap<Long, String>();
        private final java.util.Map<Long, Long> ledgerRelationStatements = new java.util.TreeMap<Long, Long>();
        private final java.util.Map<Long, Long> relationPublishers = new java.util.TreeMap<Long, Long>();
        private final java.util.Set<Long> ledgerGraphed = new java.util.TreeSet<Long>();
        private gudusoft.gsqlparser.lineage.v2.shape.Shape shape;
        /** Tests only: run on the analysis thread before dispose, with the live objects still at hand. */
        java.util.function.Consumer<gudusoft.gsqlparser.lineage.v2.shape.ShapeBuilder> shapeCheck;
        /** Tests only: build the document on a thread of its own (it reads the shape, never the analysis). */
        boolean buildOnAnotherThread;

        /** The shape the document was built from (plan 09, slice 1b). */
        gudusoft.gsqlparser.lineage.v2.shape.Shape getShape() {
            return shape;
        }
        private final java.util.Map<Long, List<Long>> ledgerEventEdges = new java.util.TreeMap<Long, List<Long>>();
        private final java.util.Map<Long, List<Long>> ledgerEventDiagnostics =
                new java.util.TreeMap<Long, List<Long>>();
        private final java.util.Map<Long, List<Long>> ledgerEventNodes = new java.util.TreeMap<Long, List<Long>>();

        /** Shadow ledger: event sequence -> nodes it stands for without an edge (a merged hop's field). */
        java.util.Map<Long, List<Long>> getLedgerEventNodes() {
            return Collections.unmodifiableMap(ledgerEventNodes);
        }

        /** Shadow ledger: event sequence -> refs of the micro-edges published for it. */
        java.util.Map<Long, List<Long>> getLedgerEventEdges() {
            return Collections.unmodifiableMap(ledgerEventEdges);
        }

        /** Shadow ledger: event sequence -> diagnostics of the UNKNOWN_INPUTs published for it. */
        java.util.Map<Long, List<Long>> getLedgerEventDiagnostics() {
            return Collections.unmodifiableMap(ledgerEventDiagnostics);
        }

        /** Shadow ledger: the refs of the statements a statement graph was built for. */
        java.util.Set<Long> getLedgerGraphed() {
            return Collections.unmodifiableSet(ledgerGraphed);
        }

        /** The value snapshot of the last analysis (plan 09, Section 3.1), or null when it could not be built. */
        gudusoft.gsqlparser.dlineage.dataflow.model.SqllrSnapshot getSnapshot() {
            return snapshot;
        }

        String getSnapshotFailure() {
            return snapshotFailure;
        }

        /** Shadow ledger (plan 09, slice 1a): relationship model id -> how the claiming statement graph handled it. */
        java.util.Map<Long, String> getLedgerRelations() {
            return Collections.unmodifiableMap(ledgerRelations);
        }

        /** Shadow ledger: relationship model id -> the statement whose graph claimed it. */
        java.util.Map<Long, Long> getLedgerRelationStatements() {
            return Collections.unmodifiableMap(ledgerRelationStatements);
        }

        /** Shadow ledger: element event sequence -> outcome. */
        java.util.Map<Long, String> getLedgerElementEvents() {
            return Collections.unmodifiableMap(ledgerElementEvents);
        }

        /** Shadow ledger: reference event sequence -> outcome. */
        java.util.Map<Long, String> getLedgerReferenceEvents() {
            return Collections.unmodifiableMap(ledgerReferenceEvents);
        }

        /**
         * Shadow ledger: frozen relationship model id -> the ref of the published statement its target event maps to
         * (the event {@code StatementGraph.valueEdges} reads, through {@code DocumentBuilder.publisher}), or -1 when no
         * published statement encloses it; absent when the relationship has no recorded target.
         */
        java.util.Map<Long, Long> getRelationPublishers() {
            return Collections.unmodifiableMap(relationPublishers);
        }

        /** Source unit ref -> the decoded text its positions are offsets into (inputs and derived units). */
        java.util.Map<Long, String> getUnitTexts() {
            return Collections.unmodifiableMap(unitTexts);
        }

        /**
         * Legacy relationship id -> the ref of the published statement that owns its target (the statement the
         * analysis recorded for the relationship, or the nearest published one enclosing it). A relationship with no
         * recorded target or no published owner is absent. This is the recorded ownership the coverage census uses
         * (plan 08, Section 14), never a position guess.
         */
        java.util.Map<Long, Long> getRelationOwners() {
            return Collections.unmodifiableMap(relationOwners);
        }

        /** The texts of the inputs, in input order. */
        List<String> getInputTexts() {
            return Collections.unmodifiableList(inputTexts);
        }

        /** The texts of the last analysis: the input texts with every statement that did not parse blanked. */
        List<String> getAnalysedTexts() {
            return Collections.unmodifiableList(analysedTexts);
        }

        /** {unit index, UTF-16 start, UTF-16 end} of each blanked statement, in the order they were found. */
        List<int[]> getBlanked() {
            return Collections.unmodifiableList(blanked);
        }

        int getAnalyses() {
            return analyses;
        }

        /** The legacy XML of the last analysis, serialized on the analysis thread before {@code dispose()}. */
        String getLegacyXml() {
            return legacyXml;
        }

        /** The legacy JSON model of the last analysis, serialized like {@link #getLegacyXml()}. */
        String getLegacyJson() {
            return legacyJson;
        }

        ProducerResult getResult() {
            return result;
        }
    }

    /** {@code capture}, when given, is filled from the analysis that published the document. */
    static ProducerResult produce(List<SqllrInput> inputs, ProducerConfig config,
            java.util.function.Consumer<SqllrObservations> prepare, Capture capture) {
        if (inputs == null || inputs.isEmpty()) {
            throw new SqllrProducerException("PRODUCER_INVALID_INPUT", "no input");
        }
        if (config == null) {
            throw new IllegalArgumentException("no config");
        }
        Option option = ProducerOptions.option(config);
        String[] texts = new String[inputs.size()];
        for (int i = 0; i < texts.length; i++) {
            String text = inputs.get(i).getText();
            if (MetadataReader.isMetadata(text) || MetadataReader.isGrabit(text) || MetadataReader.isSqlflow(text)
                    || MetadataReader.isSqlflowSharded(text)) {
                // the analyzer would read the input as a metadata export, not as SQL text the document can cite
                throw new SqllrProducerException("PRODUCER_INVALID_INPUT", "input " + i
                        + " is a metadata export, not SQL text; give metadata through ProducerConfig.metadata");
            }
            texts[i] = text;
        }
        // A statement with a syntax error leaves its whole unit unresolved; analyse again with it blanked (plan 06,
        // Section 3.8). The rerun happens only when a statement failed to parse.
        List<ParseFailures.Failed> failed = new ArrayList<ParseFailures.Failed>();
        gudusoft.gsqlparser.lineage.v2.shape.Shape projected = null;
        IdentityPolicy policy = null;
        int analyses = 0;
        for (int analysis = 1; projected == null; analysis++) {
            DataFlowAnalyzer analyzer = new DataFlowAnalyzer(infos(inputs, texts), option);
            SqllrObservations observations = new SqllrObservations(true);
            try {
                clearThreadState();
                // Always a private catalog: without one the analyzer reads the DDL of every input into its catalog
                // before any statement is resolved, so a CREATE after a statement would decide how the statement's
                // names resolve. The parse registers each declaration where it stands instead.
                analyzer.setSqlEnv(config.getMetadata() != null
                        ? MetadataSnapshot.copy(config.getMetadata().getCatalog(), config)
                        : MetadataSnapshot.empty(config.getVendor()));
                analyzer.setSqllrObservations(observations);
                if (prepare != null) {
                    prepare.accept(observations);
                }
                // what the producer reads is projected when the analysis freezes (plan 09, slice 1b): the recorder,
                // and the producer's own copy of the supplied catalog
                final gudusoft.gsqlparser.lineage.v2.shape.ShapeBuilder[] shape =
                        new gudusoft.gsqlparser.lineage.v2.shape.ShapeBuilder[1];
                final gudusoft.gsqlparser.sqlenv.TSQLEnv supplied = config.getMetadata() == null ? null
                        : MetadataSnapshot.copy(config.getMetadata().getCatalog(), config);
                observations.setFreezeHook(o -> shape[0] =
                        gudusoft.gsqlparser.lineage.v2.shape.ShapeBuilder.freeze(o, supplied));
                analyzer.generateDataFlow();
                checkTexts(texts, observations);
                if (shape[0] == null) {
                    throw new IllegalStateException("SQL-LR: the analysis was not projected: "
                            + observations.getFreezeHookFailure(), observations.getFreezeHookFailure());
                }
                List<ParseFailures.Failed> found = analysis < ParseFailures.MAX_ANALYSES
                        ? ParseFailures.collect(shape[0].shape().observations(), texts)
                        : Collections.<ParseFailures.Failed>emptyList();
                if (found.isEmpty()) {
                    // the renderings are taken now, after the analysis and before dispose, as the producer took them
                    boolean keep = capture != null && capture.shapeCheck != null;
                    projected = shape[0].complete(keep);
                    policy = IdentityPolicy.of(config.getVendor(), analyzer.getIdentifierProfile());
                    analyses = analysis;
                    if (capture != null) {
                        capture(capture, inputs, texts, failed, analysis, analyzer, config, observations);
                        capture.shape = projected;
                        if (keep) {
                            capture.shapeCheck.accept(shape[0]);
                        }
                    }
                    shape[0].release();
                } else {
                    failed.addAll(found);
                    texts = ParseFailures.blank(texts, found);
                }
            } finally {
                clearThreadState();
                observations.close();
                analyzer.dispose();
            }
        }
        // the document is built from the shape alone: the analysis is disposed
        DocumentBuilder builder = new DocumentBuilder(config, policy, inputs, projected, failed);
        if (analyses > 1) {
            builder.notes().add("analysed " + analyses + " times: " + failed.size()
                    + " statement(s) that did not parse were blanked");
        }
        ProducerResult result = capture != null && capture.buildOnAnotherThread ? buildElsewhere(builder)
                : builder.build();
        if (capture != null) {
            capture.result = result;
            owners(capture, projected.observations(), builder);
        }
        return result;
    }

    /** Tests only: {@link DocumentBuilder#build()} on a thread of its own, its failure rethrown here. */
    private static ProducerResult buildElsewhere(final DocumentBuilder builder) {
        final ProducerResult[] result = new ProducerResult[1];
        final Throwable[] failure = new Throwable[1];
        Thread t = new Thread(() -> {
            try {
                result[0] = builder.build();
            } catch (Throwable e) {
                failure[0] = e;
            }
        }, "sqllr-document");
        t.start();
        try {
            t.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("SQL-LR: interrupted while the document was built", e);
        }
        if (failure[0] instanceof RuntimeException) {
            throw (RuntimeException) failure[0];
        }
        if (failure[0] instanceof Error) {
            throw (Error) failure[0];
        }
        return result[0];
    }

    /** Runs on the analysis thread, before {@code dispose()}: the legacy model still has its thread context. */
    private static void capture(Capture capture, List<SqllrInput> inputs, String[] texts,
            List<ParseFailures.Failed> failed, int analysis, DataFlowAnalyzer analyzer, ProducerConfig config,
            SqllrObservations observations) {
        for (SqllrInput input : inputs) {
            capture.inputTexts.add(input.getText());
        }
        Collections.addAll(capture.analysedTexts, texts);
        for (ParseFailures.Failed f : failed) {
            capture.blanked.add(new int[] {f.getUnitIndex(), f.getStart(), f.getEnd()});
        }
        capture.analyses = analysis;
        gudusoft.gsqlparser.dlineage.dataflow.model.xml.dataflow legacy = analyzer.getDataFlow();
        if (legacy != null) {
            try {
                capture.legacyXml = gudusoft.gsqlparser.dlineage.util.XML2Model.saveXML(legacy);
            } catch (Exception e) {
                throw new IllegalStateException("SQL-LR capture: the legacy model did not serialize", e);
            }
            capture.legacyJson = gudusoft.gsqlparser.util.json.JSON.toJSONString(
                    DataFlowAnalyzer.getSqlflowJSONModel(legacy, config.getVendor()));
        }
        capture.snapshot = observations.getSnapshot();
        capture.snapshotFailure = observations.getSnapshotFailure();
    }

    /** What the builder decided, read from the shape it was built from. */
    private static void owners(Capture capture, gudusoft.gsqlparser.lineage.v2.shape.SqllrObservations projected,
            DocumentBuilder builder) {
        capture.unitTexts.putAll(builder.unitTexts);
        capture.ledgerRelations.putAll(builder.ledger().getRelations());
        capture.ledgerRelationStatements.putAll(builder.ledger().getRelationStatements());
        capture.ledgerElementEvents.putAll(builder.ledger().getElementEvents());
        capture.ledgerReferenceEvents.putAll(builder.ledger().getReferenceEvents());
        capture.ledgerGraphed.addAll(builder.ledger().getGraphed());
        capture.ledgerEventEdges.putAll(builder.ledger().getEventEdges());
        capture.ledgerEventDiagnostics.putAll(builder.ledger().getEventDiagnostics());
        capture.ledgerEventNodes.putAll(builder.ledger().getEventNodes());
        // the builder knows the mirrors (plan 09, slice 1b)
        if (projected.getFrozenRelations() != null) {
            for (gudusoft.gsqlparser.lineage.v2.shape.SqllrObservations.FrozenRelation f
                    : projected.getFrozenRelations()) {
                gudusoft.gsqlparser.lineage.v2.shape.SqllrObservations.ElementRecord target =
                        builder.targetOf(f.getRelation());
                if (target != null) {
                    DocumentBuilder.Stmt owner = builder.publisher(target.getStatement());
                    capture.relationPublishers.put(Long.valueOf(f.getRelation().getId()),
                            Long.valueOf(owner == null ? -1 : owner.ref));
                }
            }
        }
        for (gudusoft.gsqlparser.lineage.v2.shape.SqllrObservations.ElementRecord record : projected.getElements()) {
            if (!record.isTarget() || record.getRelation() == null) {
                continue;
            }
            DocumentBuilder.Stmt owner = builder.publisher(record.getStatement());
            if (owner != null && !capture.relationOwners.containsKey(record.getRelation().getId())) {
                capture.relationOwners.put(record.getRelation().getId(), owner.ref);
            }
        }
    }

    private static SqlInfo[] infos(List<SqllrInput> inputs, String[] texts) {
        SqlInfo[] infos = new SqlInfo[texts.length];
        for (int i = 0; i < infos.length; i++) {
            SqllrInput input = inputs.get(i);
            SqlInfo info = new SqlInfo();
            info.setSql(texts[i]);
            // The analyzer reads the file a SqlInfo names when its text is empty or white space only (an empty input,
            // or a unit whose statements were all blanked). A unit is its text: its logical name is an identity, never
            // a path to read, so such a unit gets no name (plan 09, slice 0).
            info.setFileName(gudusoft.gsqlparser.util.SQLUtil.isEmpty(texts[i]) ? null
                    : input.getLogicalName() != null ? input.getLogicalName() : "input-" + i);
            infos[i] = info;
        }
        return infos;
    }

    /**
     * Analysis state that {@code ModelBindingManager.remove()} leaves on the thread (dlineage CLAUDE.md, Concurrency):
     * an earlier analysis on this thread, or the previous analysis of this production, must not reach this one.
     */
    private static void clearThreadState() {
        ModelBindingManager.removeGlobalOraclePackage();
        ModelBindingManager.setSyntheticStmtContext(false);
        if (ModelBindingManager.get() != null) {
            ModelBindingManager.removeGlobalProcedure();
        }
    }

    /** Every published position is an offset into the input text: the analysis must have parsed exactly that text. */
    private static void checkTexts(String[] texts, SqllrObservations observations) {
        for (SqllrObservations.UnitRecord unit : observations.getUnits()) {
            int i = unit.getUnitIndex();
            if (i < 0 || i >= texts.length || !texts[i].equals(unit.getParsedText())) {
                throw new IllegalStateException("SQL-LR: the analysis parsed a text that is not input " + i);
            }
        }
    }
}