// Parse a dataflow.json file produced by the SQL compiler using the --dataflow flag,
// and create a (reduced) representation of the program representation

import { Comparable } from "./util.js";

/** Serialized JSON representation of a Dataflow graph coming from the SQL compiler. */
export interface Dataflow {
    calcite_plan: any;
    mir: MirNodes;
    sources?: Array<string>; // This field is generated by SQL-to-DBSP compiler, but is not included in the dataflow file from the support bundle. We should fall back to pipeline.program_code in the latter case
}

export interface MirNodes {
    [key: string]: MirNode | NestedMirNode;
}

export interface OutputPort {
    node: string;
    port: number;
}

export interface MirNode {
    operation: string;
    table: string | null;
    view: string | null;
    inputs: Array<OutputPort>;
    // We don't care about this yet
    calcite: any;
    positions: Array<JsonPositionRange>;
    persistent_id?: string;
}

export interface NestedNodeFixed {
    operation: "nested";
    outputs: Array<OutputPort>;
}

type NestedMirNode = NestedNodeFixed & {
    [key: string]: MirNode;
}

export interface OutputPort {
    node: string,
    output: number,
}

/** Represents the (SQL) sources of a program. */
export class Sources {
    constructor(readonly lines: Array<string>) {
        this.lines = lines;
    }

    static prefix(line: number): string {
        return line.toString().padStart(5, " ") + "| ";
    }

    static truncate(s: string): string {
        if (s.length > 100) {
            return s.substring(0, 97) + "...";
        }
        return s;
    }

    getLine(lineNo: number): string {
        // Sources may be missing
        return this.lines[lineNo] || "";
    }

    /** Given a position range, generate a string with the relevant source fragment.
     * Currently only full lines are generated, ignoring the column information. */
    public toStringFromPositionRange(range: SourcePositionRange): string {
        if (this.lines.length === 0) {
            return "";
        }
        if (range.is_empty()) return "";
        let result = "";
        for (let line = range.start.line; line <= range.end.line; line++) {
            let l = Sources.truncate(this.getLine(line - 1));
            result += Sources.prefix(line) + l + "\n";
        }
        return result;
    }

    /** Given multiple position ranges, generate the relevant source fragment.
     * Since the source fragment includes only full lines, we do not really currently
     * use the column information in any way. */
    public toString(positions: SourcePositionRanges): string {
        if (this.lines.length === 0) {
            return "";
        }
        let result = "";
        let lastLine = 0;
        for (const range of positions.sort()) {
            let nonOverlap = range.after(lastLine);
            result += this.toStringFromPositionRange(nonOverlap);
            lastLine = range.end.line;
        }
        return result;
    }
}

/** Zero or more SourcePositionRange values. */
export class SourcePositionRanges {
    constructor(readonly positions: Array<SourcePositionRange>) { }

    static empty(): SourcePositionRanges {
        return new SourcePositionRanges(new Array<SourcePositionRange>());
    }

    /** Return the source positions ranges sorted and with duplicates eliminated. */
    sort(): Array<SourcePositionRange> {
        this.positions.sort((a, b) => a.compareTo(b));
        if (this.positions.length > 1) {
            let write = 1;
            for (let read = 1; read < this.positions.length; read++) {
                if (!this.positions[read]!.equals(this.positions[read - 1]!)) {
                    this.positions[write] = this.positions[read]!;
                    write++;
                }
            }
            this.positions.length = write; // truncate
        }
        return this.positions;
    }

    // Append includes deduplication
    append(other: SourcePositionRanges) {
        for (const position of other.positions) {
            if (!this.positions.find(v => v.equals(position))) {
                this.positions.push(position);
            }
        }
    }
}

interface JsonPositionRange {
    start_line_number: number;
    start_column: number;
    end_line_number: number;
    end_column: number;
}

/** A position in the source code; lines and columns are numbered from 1. */
export class SourcePosition implements Comparable<SourcePosition> {
    constructor(readonly line: number, readonly column: number) { }

    public compareTo(other: SourcePosition): number {
        let c = Comparable.compareTo(this.line, other.line);
        if (c !== 0) return c;
        return Comparable.compareTo(this.column, other.column);
    }
}

/** A range between two source positions; the second one must be after the first one. */
export class SourcePositionRange implements Comparable<SourcePositionRange> {
    constructor(readonly range: JsonPositionRange) { }

    static empty(): SourcePositionRange {
        return new SourcePositionRange({ start_line_number: 0, start_column: 0, end_line_number: 0, end_column: 0 });
    }

    public get start(): SourcePosition {
        return new SourcePosition(this.range.start_line_number, this.range.start_column);
    }

    public get end(): SourcePosition {
        return new SourcePosition(this.range.end_line_number, this.range.end_column);
    }

    /** Compare two ranges; comparison is done lexicographically on start, end. */
    public compareTo(other: SourcePositionRange): number {
        let c = this.start.compareTo(other.start);
        if (c != 0) {
            return c;
        }
        return this.end.compareTo(other.end);
    }

    public is_empty(): boolean {
        return this.start.line === 0 && this.start.column === 0 && this.end.line === 0 && this.end.column === 0;
    }

    public equals(position: SourcePositionRange): unknown {
        return this.range.start_column === position.range.start_column &&
            this.range.start_line_number === position.range.start_line_number &&
            this.range.end_line_number === position.range.end_line_number &&
            this.range.end_column === position.range.end_column;
    }

    /** Return a range containing only the lines strictly after the specified line. */
    public after(line: number): SourcePositionRange {
        if (line >= this.end.line) {
            return SourcePositionRange.empty();
        }
        let start = this.start;
        if (start.line <= line) {
            start = new SourcePosition(line, 1);
        }
        return new SourcePositionRange({
            start_line_number: start.line,
            start_column: start.column,
            end_line_number: this.end.line,
            end_column: this.end.column
        });
    }
}