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;

    { TPipeSource acts as a selective accumulator for a single field of a stream. }
    TPipeSource = class(TInterfacedObject, ISeries)
        type
            TSharedSourceKey = record
                SourceID: Pointer;
                FieldID: Integer;
            end;

            TSharedSource = class(TObject)
            strict private
                FField: IKeyword;
                FStream: IStream;
                FTag: TSubscriptionTag;
                FObservers: TMycNotifyList<TPipeSource>;
                FFieldIndex: Integer;
                FMaxLookback: Int64;
                FData: IWriteableSeries;

                procedure HandleSignal(const Signal: TStreamSignal);

            private
                function Attach(const Observer: TPipeSource): TSubscriptionTag;
                function Detach(Tag: TSubscriptionTag): Boolean;

            public
                constructor Create(const AStream: IStream; const AField: IKeyword);
                destructor Destroy; override;
                property Field: IKeyword read FField;
                property Stream: IStream read FStream;
                property Data: IWriteableSeries read FData;
            end;

    strict private
        class var
            FCacheLock: TSpinLock;
            FCache: TObjectDictionary<TSharedSourceKey, TSharedSource>;

        class constructor CreateClass;
        class destructor DestroyClass;
    private
        FSource: TSharedSource;
        FOnSignal: TSignalProc;
        FLastSeenCycle: Int64;
        FMaxLookback: Int64;
        FTag: TSubscriptionTag;

        { 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;
        property MaxLookback: Int64 read FMaxLookback;
    end;

    TPipeConfig = TArray<TArray<TScalarRecordField>>;

    // Synchronizes multiple inputs via barrier and executes a transformation.
    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;

{ TSharedSource }

constructor TPipeSource.TSharedSource.Create(const AStream: IStream; const AField: IKeyword);
begin
    inherited Create;
    FStream := AStream;
    FField := AField;

    FMaxLookback := -1;

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

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

    FTag := FStream.Subscribe(HandleSignal);
end;

destructor TPipeSource.TSharedSource.Destroy;
begin
    FStream.Unsubscribe(FTag);
    inherited;
end;

function TPipeSource.TSharedSource.Attach(const Observer: TPipeSource): TSubscriptionTag;
begin
    FObservers.Lock;
    try
        FMaxLookback := -1;
        Result := FObservers.Advise(Observer);
    finally
        FObservers.Release;
    end;
end;

function TPipeSource.TSharedSource.Detach(Tag: TSubscriptionTag): Boolean;
begin
    FObservers.Lock;
    try
        FObservers.Unadvise(Tag);
        FMaxLookback := -1;

        // Hint to caller, if this was the last observer
        Result := FObservers.First = nil;
    finally
        FObservers.Release;
    end;
end;

procedure TPipeSource.TSharedSource.HandleSignal(const Signal: TStreamSignal);
begin
    if (Signal.Kind = skData) then
    begin
        FObservers.Lock;
        try
            if FMaxLookback < 0 then
            begin
                FMaxLookback := 0;
                var Item := FObservers.First;
                while Item <> nil do
                begin
                    var lb := Item.Receiver.MaxLookback;
                    if lb > FMaxLookback then
                        FMaxLookback := lb;
                    Item := Item.Next;
                end;
            end;

            FData.Add(Signal.Fields[FFieldIndex], FMaxLookback);

            FObservers.Notify(
                function(const Obs: TPipeSource): Boolean
                begin
                    Obs.HandleSignal(Signal);
                    Result := True;
                end
            );
        finally
            FObservers.Release;
        end;
    end;
end;

{ TPipeSource }

constructor TPipeSource.Create(const ASource: IStream; const AField: IKeyword; AMaxLookback: Int64; const AOnSignal: TSignalProc);
var
    Key: TSharedSourceKey;
begin
    inherited Create;
    FOnSignal := AOnSignal;
    FMaxLookback := AMaxLookback;
    FLastSeenCycle := -1;

    Key.SourceID := Pointer(ASource);
    Key.FieldID := AField.Idx;

    FCacheLock.Enter;
    try
        if not FCache.TryGetValue(Key, FSource) then
        begin
            FSource := TSharedSource.Create(ASource, AField);
            FCache.Add(Key, FSource);
        end;
        FTag := FSource.Attach(Self);
    finally
        FCacheLock.Exit;
    end;
end;

destructor TPipeSource.Destroy;
begin
    Detach;

    FCacheLock.Enter;
    try
        if FSource.Detach(FTag) then
        begin
            var Key: TSharedSourceKey;
            Key.SourceID := Pointer(FSource.Stream);
            Key.FieldID := FSource.Field.Idx;
            FCache.Remove(Key);
        end;
    finally
        FCacheLock.Exit;
    end;

    inherited;
end;

class constructor TPipeSource.CreateClass;
begin
    FCache := TObjectDictionary<TSharedSourceKey, TSharedSource>.Create([doOwnsValues]);
end;

class destructor TPipeSource.DestroyClass;
begin
    Assert(FCache.Count = 0, 'Not all pipes have been properly released.');
    FCache.Free;
end;

procedure TPipeSource.Detach;
begin
    FOnSignal := nil;
end;

procedure TPipeSource.HandleSignal(const Signal: TStreamSignal);
begin
    if (Signal.Kind = skData) then
    begin
        FLastSeenCycle := Signal.CycleID;
        if Assigned(FOnSignal) then
            FOnSignal(Signal);
    end;
end;

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

function TPipeSource.GetItems(Idx: Integer): TScalar;
begin
    // Test the max lookback of THIS source, because the shared one might have a larger max lookback
    if Idx >= FMaxLookback then
        raise EArgumentException.Create('Item index is greater than max lookback');

    Result := FSource.Data[Idx];
end;

function TPipeSource.GetTotalCount: Int64;
begin
    Result := FSource.Data.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.