package runtime

import (
	"context"
	"fmt"
	"time"

	"goflink/core/operator"
	"goflink/transport"
)

// RunTask runs a stream task in the background. The returned channel yields the
// first error that stopped it (context.Canceled when ctx ends early) and is
// closed when the task is done, so callers can join on it.
func RunTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], op operator.Operator[IN, OUT]) <-chan error {
	done := make(chan error, 1)

	go func() {
		defer close(done)
		if err := runTask(ctx, input, output, op); err != nil {
			done <- err
		}
	}()

	return done
}

func runTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], op operator.Operator[IN, OUT]) (err error) {
	// ponytail: assumes one writer per output channel. Fan-in needs a refcount
	// or a dedicated closer, otherwise the second Close panics.
	defer func() {
		if cerr := output.Close(); cerr != nil && err == nil {
			err = fmt.Errorf("close output: %w", cerr)
		}
	}()

	if err := op.Open(ctx); err != nil {
		return fmt.Errorf("open operator: %w", err)
	}
	defer func() {
		// ctx is usually already cancelled by the time we get here — that is
		// why the task stopped. Close still needs a live window to flush, so
		// give it one that keeps ctx's values but drops the cancellation.
		closeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
		defer cancel()

		if cerr := op.Close(closeCtx); cerr != nil && err == nil {
			err = fmt.Errorf("close operator: %w", cerr)
		}
	}()

	for {
		record, ok := input.Receive(ctx)
		if !ok {
			break
		}

		if err := op.Process(ctx, record, output); err != nil {
			return fmt.Errorf("process record: %w", err)
		}
	}

	// Receive also returns !ok on cancellation, so report that as an error
	// rather than a clean drain.
	return ctx.Err()
}