(*
; Record-Streams sind Record-Series, an die "Pipes" andocken können.
; Es handelt sich um ein reaktives Producer-Consumer-Pattern.

;; Beispiel: SMA auf Close
(def btc (create-ticker "BTCUSD"))
(def sma (create-sma 20))

; TRANSFORMATION
; btc-sma ist ein Record-Stream, der {:ma ..} records enthält.
(def btc-sma
  (pipe [btc [:Close]]
    (fn [price]
      ; KEIN emit mehr. Wir geben einfach die Map zurück.
      ; Die Engine nimmt diesen Rückgabewert und publiziert ihn.
      {:ma (sma (get price 0))}
    )
  )
)

; KOMBINATION & LOGIK
; Diese pipe hat zwei Streams als Input.
(def btc-signal
  (pipe [btc [:Close] btc-sma [:ma]]
    (fn [price ma]
        ; Hier wird einfach der Record zurückgegeben.
        ; Da der ternäre Operator (?) immer ein Ergebnis hat (:buy oder :sell),
        ; feuert dieser Stream bei jedem Tick ein Signal.
        {:signal (? (> (get price 0) (get ma 0)) :buy :sell)}
    )
  )
)

(def btc-buy-only
  (pipe [btc [:Close] btc-sma [:ma]]
    (fn [price ma]
        ; Ein IF ohne ELSE.
        ; Wenn die Bedingung FALSE ist, ist das Ergebnis der Funktion "Void".
        ; Die Engine erkennt "Void" und sendet NICHTS an die Observer.
        (if (> (get price 0) (get ma 0))
            {:signal :buy})
    )
  )
)
*)

unit Myc.Data.Stream.Pipes;

interface

uses
    System.SysUtils,
    System.Classes,
    System.Generics.Collections,
    System.SyncObjs,
    Myc.Data.Scalar,
    Myc.Data.Keyword,
    Myc.Data.Stream,
    Myc.Core.Notifier;

type
    // Forward declaration
    TPipeStream = class;

    // mapped fields per input stream
    TPipeConfig = TArray<TArray<TScalarRecordField>>;

    TPipeSource = class(TInterfacedObject, IStreamObserver)
    private
        [Weak]
        FOwner: TPipeStream;
        FSource: IStream;
        FTag: TSubscriptionTag;

        // State for barrier synchronization
        FLastSeenCycle: Int64;

        procedure OnSignal(const Signal: TStreamSignal);
        procedure Destroying;

    public
        constructor Create(AOwner: TPipeStream; ASource: IStream; const AInputs: TArray<TScalarRecordField>);
        property LastSeenCycle: Int64 read FLastSeenCycle;
    end;

    // --- Runtime: The Pipe ---
    // The executable instance of a pipe.
    // Acts as IStream (Producer) for downstream consumers and manages internal inputs.
    TPipeStream = class(TInterfacedObject, IStream)
        type
            // Functional lambda: Maps inputs (Series) to a result value (TArray<TScalar.TValue>).
            // Returns false to filter (emit nothing).
            TPipeLambda = reference to function(const Sources: array of ISeries; out Results: array of TScalar.TValue): Boolean;
    private
        FSources: TArray<TPipeSource>;
        FObservers: TMycNotifyList<IStreamObserver>;
        FSourceSeries: TArray<ISeries>;
        FSeries: IWriteableScalarRecordSeries;

        // Barrier State
        FLastFiredCycleID: Int64;

        // Execution Logic
        FLambda: TPipeLambda;

        procedure CheckBarrierAndFire(CurrentCycle: Int64);
        function GetSeries: IScalarRecordSeries;

        // Extract values from the lambda result and map them to the stream definition
        procedure Emit(const Value: TArray<TScalar.TValue>);

    public
        constructor Create(
            const AConfig: TPipeConfig;
            const ADef: IScalarRecordDefinition;
            const ASources: TArray<IStream>;
            const ALambda: TPipeLambda
        );
        destructor Destroy; override;

        // IStream Implementation
        function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
        procedure Unsubscribe(Tag: TSubscriptionTag);
    end;

implementation

{ TPipeSource }

constructor TPipeSource.Create(AOwner: TPipeStream; ASource: IStream; const AInputs: TArray<TScalarRecordField>);
begin
    inherited Create;

    FOwner := AOwner;
    FSource := ASource;
    FLastSeenCycle := -1;

    FTag := FSource.Subscribe(Self);
end;

procedure TPipeSource.Destroying;
begin
    FSource.Unsubscribe(FTag);
    FOwner := nil;
end;

procedure TPipeSource.OnSignal(const Signal: TStreamSignal);
begin
    FLastSeenCycle := Signal.CycleID;

    if Assigned(FOwner) then
        FOwner.CheckBarrierAndFire(FLastSeenCycle);
end;

{ TPipeStream }

constructor TPipeStream.Create(
    const AConfig: TPipeConfig;
    const ADef: IScalarRecordDefinition;
    const ASources: TArray<IStream>;
    const ALambda: TPipeLambda
);
begin
    inherited Create;
    FLambda := ALambda;
    FLastFiredCycleID := -1;

    Assert(Length(AConfig) = Length(ASources));

    FSeries := TScalarRecordSeries.Create(ADef);

    // Construct record series definition from source config
    SetLength(FSources, Length(ASources));
    SetLength(FSourceSeries, Length(ASources));

    var n := 0;
    for var i := 0 to High(AConfig) do
    begin
        FSources[i] := TPipeSource.Create(Self, ASources[i], AConfig[i]);
        FSources[i]._AddRef;
        inc(n, Length(AConfig[i]));
    end;

    SetLength(FSourceSeries, n);
    n := 0;
    for var i := 0 to High(AConfig) do
        for var j := 0 to High(AConfig[i]) do
        begin
            FSourceSeries[n] := ASources[i].Series.Fields[AConfig[i][j].Key];
            inc(n);
        end;
end;

destructor TPipeStream.Destroy;
begin
    for var src in FSources do
    begin
        src.Destroying;
        src._Release;
    end;
    FObservers.Finalize;
    inherited;
end;

function TPipeStream.Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
begin
    FObservers.Lock;
    try
        Result := FObservers.Advise(Observer);
    finally
        FObservers.Release;
    end;
end;

procedure TPipeStream.Unsubscribe(Tag: TSubscriptionTag);
begin
    FObservers.Lock;
    try
        FObservers.Unadvise(Tag);
    finally
        FObservers.Release;
    end;
end;

procedure TPipeStream.CheckBarrierAndFire(CurrentCycle: Int64);
begin
    FObservers.Lock;
    try
        // Barrier Check
        for var src in FSources do
        begin
            if src.LastSeenCycle < CurrentCycle then
                exit;
        end;

        // --- Barrier Open ---

        if FLastFiredCycleID >= CurrentCycle then
            exit;

        FLastFiredCycleID := CurrentCycle;

        // Execute Lambda
        if Assigned(FLambda) then
        begin
            try
                var resultVal: TArray<TScalar.TValue>;
                SetLength(resultVal, FSeries.Def.Count);

                // Handle "Void" return (Filter logic)
                if not FLambda(FSourceSeries, resultVal) then
                    exit;

                Emit(resultVal);
            except
                // Error handling strategy: Propagate/Crash for now.
                raise;
            end;
        end;
    finally
        FObservers.Release;
    end;
end;

procedure TPipeStream.Emit(const Value: TArray<TScalar.TValue>);
begin
    // Add to internal series
    FSeries.Add(Value);

    // Notify downstream
    var signal := TStreamSignal.Create(skData, FLastFiredCycleID);
    FObservers.Notify(
        function(const Obs: IStreamObserver): Boolean
        begin
            Obs.OnSignal(signal);
            Result := True;
        end
    );
end;

function TPipeStream.GetSeries: IScalarRecordSeries;
begin
    Result := FSeries;
end;

end.