import "github.com/burningxflame/gx/stream"

// Event represents something that happened, e.g. order, payment, alarm, etc.
// Event is immutable and therefore concurrency-safe.
// Event is pointer internally, and therefore can be copied at almost zero cost.
type Event struct {
    // unexported fields
}

// Create an Event
func NewEvent(id Id, tm time.Time, data any) Event

// Id of Event
type Id string

// Id of Event
func (e Event) Id() Id

// Time of Event
func (e Event) Time() time.Time

// Data of Event
func (e Event) Data() any

// Return a copy of Event e, in which the data is replaced.
func (e Event) Replace(data any) Event

// Create an Event without id nor time
func NewDataEvent(data any) Event

// Source produces Events constantly.
type Source interface {
    // Initialize the Source.
    Init() error
    // Produce Events and send them to the output stream (i.e. channel out).
    // Produce should return ASAP when ctx.Done channel is closed, which usually indicates an exit signal is sent.
    Produce(ctx context.Context, out chan<- Event)
    // Receive acks from the input stream (i.e. channel ack). An ack indicates the corresponding Event is processed.
    Ack(ctx context.Context, ack <-chan Id)
}

// Sink stores Events in files, DBs, external systems, etc.
type Sink interface {
    // Initialize the Sink.
    Init() error
    // Sink Events from the input stream (i.e. channel in), and send acks to the output stream (i.e. channel ack).
    Sink(ctx context.Context, in <-chan Event, ack chan<- Id)
}

// Processor processes Events.
type Processor interface {
    // Initialize the Processor.
    Init() error
    // Process Events from the input stream (i.e. channel in), and send results to the output streams (i.e. OutStreams outs).
    Process(ctx context.Context, in <-chan Event, outs OutStreams)
}

/*
Output Streams. An Event may be output to one of the following streams:
  - Next: Events are sent to the next Processor in the pipelines of Processors.
  - Sink: Events are sent directly to Sink, ignoring all subsequent Processors in the pipelines of Processors.
  - Drop: Events are dropped, ignoring all subsequent Processors in the pipelines of Processors.
*/
type OutStreams struct {
    Next chan<- Event
    Sink chan<- Event
    Drop chan<- Id
}

// Indicate which Output stream the Event will be sent to.
type EventOut struct {
    Event Event
    Out   Out
}

// Indicate an Output stream. See also OutStreams.
type Out byte

const (
    OutNext Out = iota + 1
    OutSink
    OutDrop
)

// Stream Processing is a composite of Source, Processor and Sink.
type StreamProc struct {
    Src  Source
    Proc Processor
    Sink Sink
    // Channel Size. Defaults to the const ChanSize.
    ChanSize int
}

// Start Stream Processing
func (s *StreamProc) Run(ctx context.Context) error

// Default Channel Size
const ChanSize = 1024

import "github.com/burningxflame/gx/stream"

// Filter and/or Map.
// Call Filter for each Event from the input stream, and drop those for which Filter returns false.
// If Filter is nil, no Event is dropped.
// For those left, call Map for each Event, and send the result to the output stream Next.
// If Map is nil, the Event itself is sent to the output stream Next.
type FilterMap struct {
    Filter func(Event) bool
    Map    func(Event) Event
}

// Consume Events from the input stream, and calculate a result.
// Once the input stream is closed, send the result to the output stream Next, and drop all consumed Events.
// If Interval is specified, periodically send intermediate results and drop consumed Events.
type Reduce struct {
    Reducer  Reducer
    Interval time.Duration
}

// Like Reduce, but split the input stream into multiple streams by KeyFn, and Reduce each stream separately.
type KeyedReduce[K comparable] struct {
    KeyFn func(Event) K
    // Reducer Generator
    ReducerGen func() Reducer
    Interval   time.Duration
}

// Consume Events and calculate a result.
type Reducer interface {
    // Consume an Event and update the result.
    Add(e Event)
    // Return the result.
    Result() ReduceResult
    // Clear the result so that the Reducer can be reused.
    Clear()

    // Tag the Reducer with the specified k-v pair.
    // Tags can be retrived from ReduceResult later.
    Tag(k, v any)
    // Set the value of the pre-defined tag key.
    // Commonly used by keyed Processors such as KeyedReduce.
    TagKey(v any)
    // Set the value of the pre-defined tag window.
    // Commonly used by window Processors such as TumblingWindow, SlidingWindow.
    TagWindow(v any)
}

// Create a Reducer.
// The Reducer applies fn to ini and the first Event,
// then applies fn to that result and the second Event, and so on.
func NewReducer[R any](ini R, fn func(R, Event) R) Reducer

// Represent a Reduce result.
type ReduceResult struct {
    Result any
}

// Return the value of tag k.
func (r *ReduceResult) Tag(k any) any

// Return the value of the pre-defined tag key.
func (r *ReduceResult) TagKey() any

// Return the value of the pre-defined tag window.
func (r *ReduceResult) TagWindow() any

import "github.com/burningxflame/gx/stream/cntwin"

// Tumbling Window.
// Split the input stream into windows, and Reduce/Process each window separately.
// Either ReducerGen or ProcFn must be specified.
type TumblingWindow struct {
    // Window Size. The number of Events in a window.
    WinSize int
    // Reducer Generator
    ReducerGen func() stream.Reducer
    // Process a window
    ProcFn func(l []stream.Event) []stream.EventOut
}

// Keyed Tumbling Window.
// Like TumblingWindow, but split the input stream into multiple streams by KeyFn, and process each stream separately.
type KeyedTumblingWindow[K comparable] struct {
    KeyFn      func(stream.Event) K
    WinSize    int
    ReducerGen func() stream.Reducer
    ProcFn     func(key K, l []stream.Event) []stream.EventOut
}

// Sliding Window.
// Split the input stream into windows, and Reduce/Process each window separately.
// Either ReducerGen or ProcFn must be specified.
type SlidingWindow struct {
    // Window Size. The number of Events in a window.
    WinSize int
    // Window Slide. How frequently a window is started.
    WinSlide int
    // Reducer Generator
    ReducerGen func() stream.Reducer
    // Process a window
    ProcFn func(l []stream.Event) []stream.EventOut
}

// Keyed Sliding Window.
// Like SlidingWindow, but split the input stream into multiple streams by KeyFn, and process each stream separately.
type KeyedSlidingWindow[K comparable] struct {
    KeyFn      func(stream.Event) K
    WinSize    int
    WinSlide   int
    ReducerGen func() stream.Reducer
    ProcFn     func(key K, l []stream.Event) []stream.EventOut
}

import "github.com/burningxflame/gx/stream/timewin"

// Tumbling Window.
// Split the input stream into windows, and Reduce/Process each window separately.
// Either ReducerGen or ProcFn must be specified.
type TumblingWindow struct {
    // Window Size
    WinSize time.Duration
    // Allowed Lateness of Event
    Late time.Duration
    // Reducer Generator
    ReducerGen func() stream.Reducer
    // Process a window. win is the start of the window.
    ProcFn func(win time.Time, l []stream.Event) []stream.EventOut
}

// Keyed Tumbling Window.
// Like TumblingWindow, but split the input stream into multiple streams by KeyFn, and process each stream separately.
type KeyedTumblingWindow[K comparable] struct {
    KeyFn      func(stream.Event) K
    WinSize    time.Duration
    Late       time.Duration
    ReducerGen func() stream.Reducer
    ProcFn     func(key K, win time.Time, l []stream.Event) []stream.EventOut
}

// Sliding Window.
// Split the input stream into windows, and Reduce/Process each window separately.
// Either ReducerGen or ProcFn must be specified.
type SlidingWindow struct {
    // Window Size
    WinSize time.Duration
    // Window Slide. How frequently a window is started.
    WinSlide time.Duration
    // Allowed Lateness of Event
    Late time.Duration
    // Reducer Generator
    ReducerGen func() stream.Reducer
    // Process a window. win is the start of the window.
    ProcFn func(win time.Time, l []stream.Event) []stream.EventOut
}

// Keyed Sliding Window.
// Like SlidingWindow, but split the input stream into multiple streams by KeyFn, and process each stream separately.
type KeyedSlidingWindow[K comparable] struct {
    KeyFn      func(stream.Event) K
    WinSize    time.Duration
    WinSlide   time.Duration
    Late       time.Duration
    ReducerGen func() stream.Reducer
    ProcFn     func(key K, win time.Time, l []stream.Event) []stream.EventOut
}

// Session Window.
// Split the input stream into windows, and Reduce/Process each window separately.
// Either ReducerGen or ProcFn must be specified.
type SessionWindow struct {
    // Session Gap
    Gap time.Duration
    // Allowed Lateness of Event
    Late time.Duration
    // Reducer Generator
    ReducerGen func() stream.Reducer
    // Process a window. win is the start of the window.
    ProcFn func(win time.Time, l []stream.Event) []stream.EventOut
}

// Keyed Session Window.
// Like SessionWindow, but split the input stream into multiple streams by KeyFn, and process each stream separately.
type KeyedSessionWindow[K comparable] struct {
    KeyFn      func(stream.Event) K
    Gap        time.Duration
    Late       time.Duration
    ReducerGen func() stream.Reducer
    ProcFn     func(key K, win time.Time, l []stream.Event) []stream.EventOut
}

import "github.com/burningxflame/gx/stream/composite"

// Chain Processors.
// The output stream (Next) of the first Processor is the input stream of the second Processor, and so on.
type Chain struct {
    // The Processors to be chained. 2 at least.
    Procs []stream.Processor
    // Channel Size. Defaults to the const ChanSize.
    ChanSize int
}

// Distribute the input stream to N Processors.
// Each Processor processes a portion of the input stream.
type Distribute struct {
    // Processor Generator
    ProcGen func() stream.Processor
    // The number of Processors. 2 at least.
    N int
}

// Compound multiple Processors into one.
// Broadcast the input stream to all internal Processors, and merge the output streams.
// Every Event to be processed by Compound is processed by all internal Processors, and the results are merged.
type Compound struct {
    // The Processors to be compounded. 2 at least.
    Procs []stream.Processor
    // Merge the results of an Event.
    Merge func([]stream.EventOut) stream.EventOut
    // Channel Size. Defaults to the const ChanSize.
    ChanSize int
}

import "github.com/burningxflame/gx/stream/composite"

// Combine multiple Sources into one.
// All Event IDs must be unique, even if they are produced by different internal Sources.
type MultiSource struct {
    // The Sources to be combined. 2 at least.
    Srcs []stream.Source
    // Channel Size. Defaults to the const ChanSize.
    ChanSize int
}

import "github.com/burningxflame/gx/stream/composite"

// Combine multiple Sinks into one.
// Broadcast the input stream to all internal Sinks.
// Every Event to be sinked and acked by MultiSink is sinked and acked by all internal Sinks.
type MultiSink struct {
    // The Sinks to be combined. 2 at least.
    Sinks []stream.Sink
    // Channel Size. Defaults to the const ChanSize.
    ChanSize int
}