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
    private
        FCycleID: Int64;
        FData: TScalar.PValue;
        function GetFields(Idx: Integer): TScalar.TValue; inline;
        function GetKind: TSignalKind; inline;
    public
        constructor Create(ACycleID: Int64; const AData: TScalar.PValue);
        function ToString: string;
        property Fields[Idx: Integer]: TScalar.TValue read GetFields;
        property CycleID: Int64 read FCycleID;
        property Kind: TSignalKind read GetKind;
    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(CycleID: Int64; Values: TScalar.PValue);
    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;
        FMaxLookback: Int64;

        { ISeries implementation }
        function GetCount: Int64;
        function GetItems(Idx: Integer): TScalar;
        function GetTotalCount: Int64;
        procedure HandleSignal(const Signal: TStreamSignal);

    private
        procedure Detach;

    public
        constructor Create(const ASource: IStream; const AField: IKeyword; AMaxLookback: Int64; const AOnSignal: TSignalProc);
        destructor Destroy; override;

        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;
        FMaxLookback: Int64;
        FRequiredLookback: Int64;
        FLambda: TPipeLambda;

    private
        procedure CheckBarrierAndFire(CurrentCycle: Int64);

    public
        constructor Create(
            const AConfig: TPipeConfig;
            const ADef: IScalarRecordDefinition;
            const ASources: TArray<IStream>;
            const ALambda: TPipeLambda;
            AMaxLookback, ARequiredLookback: Int64
        );
        destructor Destroy; override;
        property MaxLookback: Int64 read FMaxLookback;
        property RequiredLookback: Int64 read FRequiredLookback;
    end;

implementation

{ TStreamSignal }

constructor TStreamSignal.Create(ACycleID: Int64; const AData: TScalar.PValue);
begin
    FCycleID := ACycleID;
    FData := AData;
end;

function TStreamSignal.GetFields(Idx: Integer): TScalar.TValue;
var
    P: TScalar.PValue;
begin
    Assert(FData <> nil, 'Heartbeat does not contain data');
    P := FData;
    inc(P, Idx);
    Result := P^;
end;

function TStreamSignal.GetKind: TSignalKind;
begin
    if FData <> nil then
        Result := skData
    else
        Result := skHeartbeat;
end;

function TStreamSignal.ToString: string;
begin
    if Kind = skData then
        Result := Format('Signal(Data, #%d)', [CycleID])
    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(CycleID: Int64; Values: TScalar.PValue);
var
    LSignal: TStreamSignal;
begin
    LSignal.Create(CycleID, Values);
    FObservers.Lock;
    try
        FObservers.Notify(
            function(const Obs: TSignalProc): Boolean
            begin
                Obs(LSignal);
                Result := True;
            end
        );
    finally
        LSignal.FData := nil;
        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(FLastCycleID, @RowData[0]);
end;

{ TPipeSource }

constructor TPipeSource.Create(const ASource: IStream; const AField: IKeyword; AMaxLookback: Int64; const AOnSignal: TSignalProc);
begin
    inherited Create;
    FSource := ASource;
    FOnSignal := AOnSignal;
    FMaxLookback := AMaxLookback;
    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.Fields[FFieldIndex], FMaxLookback);
        if Assigned(FOnSignal) then
            FOnSignal(Signal);
    end;
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;
    AMaxLookback, ARequiredLookback: Int64
);
var
    i, j, n: Integer;
begin
    inherited Create(ADef);
    FLambda := ALambda;
    FMaxLookback := AMaxLookback;
    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, ADef.Count);

    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,
                    FMaxLookback,
                    procedure(const Signal: TStreamSignal) begin CheckBarrierAndFire(Signal.CycleID); end
                );
            inc(n);
        end;
    end;
    FRequiredLookback := ARequiredLookback;
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
    if FLastFiredCycleID >= CurrentCycle then
        exit;

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

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

    FLastFiredCycleID := CurrentCycle;

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

initialization
    TMycNotifyList<TSignalProc>.ReverseOnNotify := False;

end.