package media

import "time"

type PipeLineStageEffect func([]byte) error

type PipeLineStage struct {
	effect           PipeLineStageEffect
	outputBufferSize int
}

// The pipe head channel MUST be managed by the caller. Closing the channel is the
// only reliable method to tear down the pipeline. Closing the head channel causes all
// pipeline stages to return via a cascade of channel closures. After the final stage returns,
// the pipe tail will be closed.
//
// The returned error channel should be used in a select statement with the pipe tail, if an
// error is received, the caller MUST close the input channel or the pipeline stages will keep
// spinning until the source is exhausted. An error in one of the stages will not close
// the pipe tail channel, rather the errored pipeline stage will become a sink, and the tail will
// stop receiving data after subsequent buffers are emptied.
//
// When a pipeline stage errors, it enters a "sink mode" where it continues
// to consume input data at a throttled rate but does not do work or output data.
// - Prevents deadlocks by ensuring the pipeline can drain
// - Maintains backpressure to avoid overwhelming upstream stages
// - Minimizes resource usage during error state
//
// tldr; The caller must always close the pipe head input channel to ensure a proper and timely cleanup.
func NewPipeLine(pipeHead <-chan []byte, stages ...PipeLineStage) (<-chan []byte, <-chan error) {
	var pipeTail chan []byte
	cherror := make(chan error, 1)

	// just forward to output if no stages were added.
	if len(stages) == 0 {
		pipeTail = make(chan []byte, 1)

		go func() {
			defer close(pipeTail)
			for {
				b, ok := <-pipeHead

				if !ok {
					return
				}

				pipeTail <- b
			}
		}()
		return pipeTail, cherror
	}

	in := pipeHead
	for idx, stage := range stages {
		out := make(chan []byte, stage.outputBufferSize)

		if idx+1 == len(stages) {
			pipeTail = out
		}

		go func(in <-chan []byte, out chan<- []byte, f PipeLineStageEffect) {
			defer close(out)

			for {
				buf, ok := <-in

				if !ok { // closed
					return
				}

				err := f(buf)

				if err != nil {
					cherror <- err
					break
				}

				out <- buf
			}

			// sink data here until input is closed
			// has the effect of stalling further reads from the pipe tail
			// after subsequent buffers have been emptied.
			for {
				if _, ok := <-in; !ok {
					return
				}
				time.Sleep(15 * time.Millisecond)
			}

		}(in, out, stage.effect)

		in = out
	}

	return pipeTail, cherror
}