unit Test.Myc.Ast.Stream.Pipes;

interface

uses
    DUnitX.TestFramework,
    System.SysUtils,
    System.Generics.Collections,
    // Core Units
    Myc.Data.Scalar,
    Myc.Data.Keyword,
    Myc.Data.Value,
    Myc.Data.Series,
    Myc.Data.Stream,
    Myc.Data.Stream.Pipes,
    // AST & Compiler
    Myc.Ast,
    Myc.Ast.Nodes,
    Myc.Ast.Scope,
    Myc.Ast.Types,
    Myc.Ast.Identities,
    Myc.Ast.Script,
    Myc.Ast.Compiler.Binder,
    Myc.Ast.Compiler.TypeChecker,
    Myc.Ast.Evaluator;

type
    // Helper: Captures compiler errors so we can assert on them
    TCapturingLog = class(TInterfacedObject, ICompilerLog)
    private
        FErrors: TList<string>;
    public
        constructor Create;
        destructor Destroy; override;
        procedure Add(ALevel: TCompilerErrorLevel; const AMessage: string; const ANode: IAstNode = nil);
        procedure AddError(const AMessage: string; const ANode: IAstNode = nil);
        procedure AddWarning(const AMessage: string; const ANode: IAstNode = nil);
        function HasErrors: Boolean;
        function GetEntryCount: Integer;
        function GetEntries: TArray<TCompilerError>;
        property Errors: TList<string> read FErrors;
    end;

    // Helper: Collects stream output for assertions
    TDataCollector = class(TInterfacedObject, IStreamObserver)
    private
        FStream: IStream;
        FLastRecord: IScalarRecord;
        FSignalCount: Integer;
    public
        constructor Create(AStream: IStream);
        procedure OnSignal(const Signal: TStreamSignal);
        property LastRecord: IScalarRecord read FLastRecord;
        property SignalCount: Integer read FSignalCount;
    end;

    [TestFixture]
    TPipeIntegrationTests = class
    private
        FGraph: IGraphExecutor;
        FLog: TCapturingLog;
        FRootLayout: IScopeLayout;
        FRootScope: IExecutionScope;
        FInputChannel: IInputChannel;

        // Simulates the Host Application setting up the environment ('btc' variable)
        procedure SetupHostEnvironment;
        // Compiles and Runs a script source
        function CompileAndRun(const Source: string): IStream;
    public
        [Setup]
        procedure Setup;
        [Teardown]
        procedure Teardown;

        [Test]
        procedure Test_Simple_Pipe_Throughput;

        [Test]
        procedure Test_Pipe_Field_Mapping_And_Calculation;
    end;

implementation

{ TCapturingLog }

constructor TCapturingLog.Create;
begin
    FErrors := TList<string>.Create;
end;

destructor TCapturingLog.Destroy;
begin
    FErrors.Free;
    inherited;
end;

procedure TCapturingLog.Add(ALevel: TCompilerErrorLevel; const AMessage: string; const ANode: IAstNode);
begin
    if ALevel = elError then
        FErrors.Add(AMessage);
end;

procedure TCapturingLog.AddError(const AMessage: string; const ANode: IAstNode);
begin
    Add(elError, AMessage, ANode);
end;

procedure TCapturingLog.AddWarning(const AMessage: string; const ANode: IAstNode);
begin
    // Ignore warnings for tests
end;

function TCapturingLog.GetEntries: TArray<TCompilerError>;
begin
    Result := nil;
end;

function TCapturingLog.GetEntryCount: Integer;
begin
    Result := FErrors.Count;
end;

function TCapturingLog.HasErrors: Boolean;
begin
    Result := FErrors.Count > 0;
end;

{ TDataCollector }

constructor TDataCollector.Create(AStream: IStream);
begin
    FStream := AStream;
    FSignalCount := 0;
    FLastRecord := nil;
end;

procedure TDataCollector.OnSignal(const Signal: TStreamSignal);
var
    scalarVal: TScalar;
begin
    if Signal.Kind = skData then
    begin
        Inc(FSignalCount);
        // Grab the latest item from the series
        if FStream.Series.Count > 0 then
        begin
            scalarVal := FStream.Series.Items[FStream.Series.Count - 1];
            // FIX: Cast TScalar to TDataValue to access AsScalarRecord
            FLastRecord := TDataValue(scalarVal).AsScalarRecord;
        end;
    end;
end;

{ TPipeIntegrationTests }

procedure TPipeIntegrationTests.Setup;
begin
    FGraph := TGraphExecutor.Create;
    FLog := TCapturingLog.Create;
    SetupHostEnvironment;
end;

procedure TPipeIntegrationTests.Teardown;
begin
    FGraph := nil; // Interfaces release automatically
    FLog.Free; // Class implementation needs free (ref counting mixed usage)
end;

procedure TPipeIntegrationTests.SetupHostEnvironment;
var
    btcDef: IScalarRecordDefinition;
    btcType: IStaticType;
    typeDesc: IScopeDescriptor;
    btcSlot: Integer;
    scopeBuilder: IScopeBuilder;
begin
    // 1. Create the physical Input Channel in the Graph
    // FIX: Use TKeywordRegistry.Intern for Keywords
    // FIX: Using TScalarRecordDefinition class (check if renamed to TRecordDef in your codebase)
    btcDef :=
        TScalarRecordDefinition.Create(
            [
                TScalarRecordField.Create(TKeywordRegistry.Intern('Close'), skInt64),
                TScalarRecordField.Create(TKeywordRegistry.Intern('Open'), skInt64)
            ]
        );
    FInputChannel := FGraph.GetInputChannel('btc', btcDef);

    // 2. Prepare the Compiler Environment (Symbol Table)
    // FIX: Use Builder to define symbols
    scopeBuilder := TScope.CreateBuilder(nil);
    btcSlot := scopeBuilder.Define('btc');
    FRootLayout := scopeBuilder.Build;

    // 3. Prepare Type Information
    // Tell the TypeChecker that 'btc' is a RecordSeries with Open/Close fields
    btcType := TTypes.CreateRecordSeries(btcDef);
    typeDesc := TScope.CreateDescriptor(FRootLayout, [btcType]);

    // 4. Prepare Runtime Scope
    // Create scope and inject the actual Stream object into slot 0
    FRootScope := TScope.CreateScope(nil, typeDesc, nil);
    FRootScope.DefineBoxed(btcSlot, TDataValue.FromStream(FInputChannel.Stream));
end;

function TPipeIntegrationTests.CompileAndRun(const Source: string): IStream;
var
    rawAst, boundAst, typedAst: IAstNode;
    evaluator: IEvaluatorVisitor;
    res: TDataValue;
begin
    // 1. Parse
    try
        rawAst := TAstScript.Parse(Source);
    except
        on E: Exception do
            Assert.Fail('Parsing failed: ' + E.Message);
    end;

    // 2. Bind
    boundAst := TAstBinder.Bind(FRootLayout, rawAst, FRootLayout, FLog);
    if FLog.HasErrors then
        Assert.Fail('Binding failed: ' + FLog.Errors[0]);

    // 3. TypeCheck
    typedAst := TTypeChecker.CheckTypes(boundAst, FRootLayout, FRootScope, FLog);
    if FLog.HasErrors then
        Assert.Fail('TypeCheck failed: ' + FLog.Errors[0]);

    // 4. Evaluate
    evaluator := TEvaluatorVisitor.Create(FRootScope);
    res := evaluator.Execute(typedAst);

    Assert.AreEqual(vkStream, res.Kind, 'Evaluation did not return a Stream');
    Result := res.AsStream;
end;

procedure TPipeIntegrationTests.Test_Simple_Pipe_Throughput;
var
    pipeStream: IStream;
    observer: TDataCollector;
    inputData: IKeywordMapping<TScalar>;
begin
    // A simple pipe that just passes 'Close' through as 'Val'
    var source := '(pipe [btc [:Close]] (fn [c] { :Val c }))';

    pipeStream := CompileAndRun(source);

    // Attach Observer
    observer := TDataCollector.Create(pipeStream);
    pipeStream.Subscribe(observer);

    // --- Step 1: Input 100 ---
    inputData := TKeywordMapping<TScalar>.Create([TPair<IKeyword, TScalar>.Create(TKeywordRegistry.Intern('Close'), 100)]);
    FInputChannel.Push(inputData);

    FGraph.Step;

    // Assertions
    Assert.AreEqual(1, observer.SignalCount, 'Should have received 1 signal');
    Assert.IsNotNull(observer.LastRecord, 'Record should not be nil');
    Assert.AreEqual(Int64(100), observer.LastRecord.Fields[TKeywordRegistry.Intern('Val')].Value.AsInt64, 'Output Value wrong');

    // --- Step 2: Input 200 ---
    inputData := TKeywordMapping<TScalar>.Create([TPair<IKeyword, TScalar>.Create(TKeywordRegistry.Intern('Close'), 200)]);
    FInputChannel.Push(inputData);

    FGraph.Step;

    Assert.AreEqual(2, observer.SignalCount);
    Assert.AreEqual(Int64(200), observer.LastRecord.Fields[TKeywordRegistry.Intern('Val')].Value.AsInt64);
end;

procedure TPipeIntegrationTests.Test_Pipe_Field_Mapping_And_Calculation;
var
    pipeStream: IStream;
    observer: TDataCollector;
    inputData: IKeywordMapping<TScalar>;
begin
    // A pipe that takes Open and Close, calculates the delta, and renames fields
    // fn params: o -> Open, c -> Close
    var source := '(pipe [btc [:Open :Close]] (fn [o c] { :Delta (- c o) :Original c }))';

    pipeStream := CompileAndRun(source);
    observer := TDataCollector.Create(pipeStream);
    pipeStream.Subscribe(observer);

    // Input: Open=100, Close=110 -> Delta should be 10
    inputData :=
        TKeywordMapping<TScalar>.Create(
            [
                TPair<IKeyword, TScalar>.Create(TKeywordRegistry.Intern('Open'), 100),
                TPair<IKeyword, TScalar>.Create(TKeywordRegistry.Intern('Close'), 110)
            ]
        );
    FInputChannel.Push(inputData);

    FGraph.Step;

    Assert.AreEqual(1, observer.SignalCount);

    var delta := observer.LastRecord.Fields[TKeywordRegistry.Intern('Delta')].Value.AsInt64;
    var orig := observer.LastRecord.Fields[TKeywordRegistry.Intern('Original')].Value.AsInt64;

    Assert.AreEqual(Int64(10), delta, 'Delta calculation wrong');
    Assert.AreEqual(Int64(110), orig, 'Original value passthrough wrong');
end;

initialization
    TDUnitX.RegisterTestFixture(TPipeIntegrationTests);

end.