unit Myc.Data.Stream;

interface

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

type
    TSignalKind = (skData, skHeartbeat);

    // TStreamSignal represents the data packet traveling through the graph.
    // It carries the CycleID for synchronization and the actual data payload.
    TStreamSignal = record
    strict private
        FCycleID: Int64;
        FKind: TSignalKind;
        FData: TArray<TScalar.TValue>;
    public
        constructor Create(AKind: TSignalKind; ACycleID: Int64; const AData: TArray<TScalar.TValue> = nil);
        function ToString: string;
        property CycleID: Int64 read FCycleID;
        property Kind: TSignalKind read FKind;
        property Data: TArray<TScalar.TValue> read FData;
    end;

    TSignalProc = reference to procedure(const Signal: TStreamSignal);

    TSubscriptionTag = Pointer;

    // IStream is now stateless. It defines WHAT is produced (Def), but not
    //  the history. History must be captured by an observer (Accumulator).
    IStream = interface
        {$region 'private'}
        function GetDef: IScalarRecordDefinition;
        {$endregion}

        function Subscribe(const Observer: TSignalProc): TSubscriptionTag;
        procedure Unsubscribe(Tag: TSubscriptionTag);

        { Returns the definition (schema) of the records this stream produces. }
        property Def: IScalarRecordDefinition read GetDef;
    end;

    // Implements broadcasting logic. No internal storage of past values.
    TCustomDataStream = class(TInterfacedObject, IStream)
    strict private
        FDef: IScalarRecordDefinition;
        FObservers: TMycNotifyList<TSignalProc>;
        function GetDef: IScalarRecordDefinition;
    protected
        { Broadcasts data to all subscribers. Stateless. }
        procedure Emit(const Value: array of TScalar.TValue; ACycleID: Int64);
    public
        constructor Create(const ADef: IScalarRecordDefinition);
        destructor Destroy; override;

        function Subscribe(const Observer: TSignalProc): TSubscriptionTag;
        procedure Unsubscribe(Tag: TSubscriptionTag);
        property Def: IScalarRecordDefinition read GetDef;
    end;

    // The entry point for external data. Generates the global CycleID (Clock).
    TRootStream = class(TCustomDataStream)
    strict private
        FLastCycleID: Int64;
    public
        constructor Create(const ADef: IScalarRecordDefinition);
        { Injects a new record into the stream system. }
        procedure Push(const RowData: array of TScalar.TValue);
    end;

    // Synchronizes multiple inputs via barrier and executes a transformation.
    TPipeStream = class;

    { TPipeSource acts as a selective accumulator for a single field of a stream. }
    TPipeSource = class(TInterfacedObject, ISeries)
    public
        type
            TSignalProc = reference to procedure(const Signal: TStreamSignal);
    strict private
        FSource: IStream;
        FTag: TSubscriptionTag;
        FOnSignal: TSignalProc;
        FFieldIndex: Integer;
        FData: IWriteableSeries;
        FLastSeenCycle: Int64;
        FLookback: Int64;

        { ISeries implementation }
        function GetCount: Int64;
        function GetItems(Idx: Integer): TScalar;
        function GetTotalCount: Int64;
    private
        procedure HandleSignal(const Signal: TStreamSignal);
    protected
        procedure Detach;
    public
        constructor Create(const ASource: IStream; const AField: IKeyword; ALookback: Int64; const AOnSignal: TSignalProc);
        destructor Destroy; override;

        function IsReady(RequiredLookback: Int64): Boolean;
        property LastSeenCycle: Int64 read FLastSeenCycle;
    end;

    TPipeConfig = TArray<TArray<TScalarRecordField>>;

    TPipeStream = class(TCustomDataStream)
    public
        type
            TPipeLambda = reference to function(const Sources: array of ISeries; out Results: array of TScalar.TValue): Boolean;
    strict private
        FSources: TArray<ISeries>;
        FResults: TArray<TScalar.TValue>;
        FLastFiredCycleID: Int64;
        FLookback: Int64;
        FLambda: TPipeLambda;

    private
        procedure CheckBarrierAndFire(CurrentCycle: Int64);

    public
        constructor Create(
            const AConfig: TPipeConfig;
            const ADef: IScalarRecordDefinition;
            const ASources: TArray<IStream>;
            const ALambda: TPipeLambda;
            ALookback: Int64
        );
        destructor Destroy; override;
        property Lookback: Int64 read FLookback;
    end;

implementation

{ TStreamSignal }

constructor TStreamSignal.Create(AKind: TSignalKind; ACycleID: Int64; const AData: TArray<TScalar.TValue>);
begin
    FKind := AKind;
    FCycleID := ACycleID;
    FData := AData;
end;

function TStreamSignal.ToString: string;
begin
    if Kind = skData then
        Result := Format('Signal(Data, #%d, Fields: %d)', [CycleID, Length(Data)])
    else
        Result := Format('Signal(Heartbeat, #%d)', [CycleID]);
end;

{ TCustomDataStream }

constructor TCustomDataStream.Create(const ADef: IScalarRecordDefinition);
begin
    inherited Create;
    FDef := ADef;
end;

destructor TCustomDataStream.Destroy;
begin
    FObservers.Finalize;
    inherited;
end;

function TCustomDataStream.GetDef: IScalarRecordDefinition;
begin
    Result := FDef;
end;

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

procedure TCustomDataStream.Unsubscribe(Tag: TSubscriptionTag);
begin
    if Tag = nil then
        exit;

    FObservers.Lock;
    try
        FObservers.Unadvise(Tag);
    finally
        FObservers.Release;
    end;
end;

procedure TCustomDataStream.Emit(const Value: array of TScalar.TValue; ACycleID: Int64);
var
    LSignal: TStreamSignal;
    LData: TArray<TScalar.TValue>;
    I: Integer;
begin
    // Convert open array to dynamic array for signal payload
    SetLength(LData, Length(Value));
    for I := 0 to High(Value) do
        LData[I] := Value[I];

    LSignal := TStreamSignal.Create(skData, ACycleID, LData);

    FObservers.Lock;
    try
        FObservers.Notify(
            function(const Obs: TSignalProc): Boolean
            begin
                Obs(LSignal);
                Result := True;
            end
        );
    finally
        FObservers.Release;
    end;
end;

{ TRootStream }

constructor TRootStream.Create(const ADef: IScalarRecordDefinition);
begin
    inherited Create(ADef);
    FLastCycleID := 0;
end;

procedure TRootStream.Push(const RowData: array of TScalar.TValue);
begin
    if Length(RowData) <> Def.Count then
        raise EArgumentException.Create('Push: Data field count mismatch.');

    Inc(FLastCycleID);
    Emit(RowData, FLastCycleID);
end;

{ TPipeSource }

constructor TPipeSource.Create(const ASource: IStream; const AField: IKeyword; ALookback: Int64; const AOnSignal: TSignalProc);
begin
    inherited Create;
    FSource := ASource;
    FOnSignal := AOnSignal;
    FLookback := ALookback;
    FLastSeenCycle := -1;

    FFieldIndex := ASource.Def.IndexOf(AField);
    if (FFieldIndex < 0) then
        raise EArgumentException.CreateFmt('Field %s not found in source stream.', [AField.Name]);

    FData := TScalarSeries.Create(ASource.Def.Items[FFieldIndex]);

    FTag := FSource.Subscribe(procedure(const Signal: TStreamSignal) begin HandleSignal(Signal); end);
end;

destructor TPipeSource.Destroy;
begin
    Detach;
    inherited;
end;

procedure TPipeSource.Detach;
begin
    FSource.Unsubscribe(FTag);
    FTag := nil;
    FOnSignal := nil;
end;

procedure TPipeSource.HandleSignal(const Signal: TStreamSignal);
begin
    if (Signal.Kind = skData) then
    begin
        FLastSeenCycle := Signal.CycleID;
        FData.Add(Signal.Data[FFieldIndex], FLookback);
        if Assigned(FOnSignal) then
            FOnSignal(Signal);
    end;
end;

function TPipeSource.IsReady(RequiredLookback: Int64): Boolean;
begin
    Result := (FLastSeenCycle >= 0) and (FData.Count >= RequiredLookback);
end;

function TPipeSource.GetCount: Int64;
begin
    Result := FData.Count;
end;

function TPipeSource.GetItems(Idx: Integer): TScalar;
begin
    Result := FData.Items[Idx];
end;

function TPipeSource.GetTotalCount: Int64;
begin
    Result := FData.TotalCount;
end;

{ TPipeStream }

constructor TPipeStream.Create(
    const AConfig: TPipeConfig;
    const ADef: IScalarRecordDefinition;
    const ASources: TArray<IStream>;
    const ALambda: TPipeLambda;
    ALookback: Int64
);
var
    i, j, n: Integer;
begin
    inherited Create(ADef);
    FLambda := ALambda;
    FLookback := ALookback;
    FLastFiredCycleID := -1;

    // Flatten sources from config
    n := 0;
    for i := 0 to High(AConfig) do
        inc(n, Length(AConfig[i]));

    SetLength(FSources, n);
    SetLength(FResults, n);

    n := 0;
    for i := 0 to High(ASources) do
    begin
        for j := 0 to High(AConfig[i]) do
        begin
            // Each input is wrapped in an accumulator source
            FSources[n] :=
                TPipeSource.Create(
                    ASources[i],
                    AConfig[i][j].Key,
                    FLookback,
                    procedure(const Signal: TStreamSignal) begin CheckBarrierAndFire(Signal.CycleID); end
                );
            inc(n);
        end;
    end;
end;

destructor TPipeStream.Destroy;
begin
    for var i := 0 to High(FSources) do
        (FSources[i] as TPipeSource).Detach;
    inherited;
end;

procedure TPipeStream.CheckBarrierAndFire(CurrentCycle: Int64);
begin
    const requiredLookback = 10;

    // 1. Check if all sources reached this cycle AND have enough history (Warm-up)
    for var src in FSources do
    begin
        if src.Count < requiredLookback then
            exit;

        var LSource := src as TPipeSource;
        if LSource.LastSeenCycle < CurrentCycle then
            exit;
    end;

    if FLastFiredCycleID >= CurrentCycle then
        Exit;

    FLastFiredCycleID := CurrentCycle;

    if Assigned(FLambda) then
    begin
        // Execute transformation on synchronized window
        if FLambda(FSources, FResults) then
        begin
            Emit(FResults, FLastFiredCycleID);
        end;
    end;
end;

initialization
    TMycNotifyList<TSignalProc>.ReverseOnNotify := False;

end.