uopi/internal/datasource/synthetic/timestamp_test.go

package synthetic

import (
	"context"
	"log/slog"
	"os"
	"testing"
	"time"

	"github.com/uopi/uopi/internal/broker"
	"github.com/uopi/uopi/internal/datasource"
)

// seqSource is a test DataSource that emits a fixed sequence of values, each
// carrying its own timestamp, so tests can control upstream sample times.
type seqSource struct {
	name string
	seq  []datasource.Value
}

func (s *seqSource) Name() string                                               { return s.name }
func (s *seqSource) Connect(context.Context) error                              { return nil }
func (s *seqSource) ListSignals(context.Context) ([]datasource.Metadata, error) { return nil, nil }
func (s *seqSource) GetMetadata(context.Context, string) (datasource.Metadata, error) {
	return datasource.Metadata{Name: "x", Type: datasource.TypeFloat64}, nil
}
func (s *seqSource) Write(context.Context, string, any) error { return datasource.ErrNotWritable }
func (s *seqSource) History(context.Context, string, time.Time, time.Time, int) ([]datasource.Value, error) {
	return nil, datasource.ErrHistoryUnavailable
}
func (s *seqSource) Subscribe(ctx context.Context, _ string, ch chan<- datasource.Value) (datasource.CancelFunc, error) {
	go func() {
		for _, v := range s.seq {
			select {
			case ch <- v:
			case <-ctx.Done():
				return
			}
			time.Sleep(8 * time.Millisecond)
		}
	}()
	return func() {}, nil
}

// TestSubscribePreservesUpstreamTimestamp verifies a single-source synthetic
// emits each computed value with the upstream sample's timestamp.
func TestSubscribePreservesUpstreamTimestamp(t *testing.T) {
	log := slog.New(slog.NewTextHandler(os.Stderr, nil))
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	base := time.Date(2026, 6, 19, 10, 0, 0, 0, time.UTC)
	src := &seqSource{name: "src", seq: []datasource.Value{
		{Timestamp: base.Add(1 * time.Second), Data: 1.0, Quality: datasource.QualityGood},
		{Timestamp: base.Add(2 * time.Second), Data: 2.0, Quality: datasource.QualityGood},
		{Timestamp: base.Add(3 * time.Second), Data: 3.0, Quality: datasource.QualityGood},
	}}

	brk := broker.New(ctx, log)
	brk.Register(src)
	syn := New(t.TempDir(), brk, log)
	if err := syn.Connect(ctx); err != nil {
		t.Fatal(err)
	}
	if err := syn.AddSignal(SignalDef{
		Name: "g", DS: "src", Signal: "x",
		Pipeline: []NodeDef{{Type: "gain", Params: map[string]any{"gain": 10.0}}},
	}); err != nil {
		t.Fatal(err)
	}

	ch := make(chan datasource.Value, 8)
	if _, err := syn.Subscribe(ctx, "g", ch); err != nil {
		t.Fatal(err)
	}

	want := []time.Time{base.Add(1 * time.Second), base.Add(2 * time.Second), base.Add(3 * time.Second)}
	for i, w := range want {
		select {
		case v := <-ch:
			if !v.Timestamp.Equal(w) {
				t.Errorf("emit #%d timestamp: want %s, got %s", i, w, v.Timestamp)
			}
		case <-time.After(2 * time.Second):
			t.Fatalf("timeout waiting for emit #%d", i)
		}
	}
}

// TestSubscribeMultiSourceUsesLatestTimestamp verifies that a synthetic combining
// two independent sources stamps each output with the MOST RECENT contributing
// sample time — not the timestamp of whichever source happened to trigger the
// computation. A slow source carrying a stale timestamp must not drag the output
// backwards in time (which previously produced wrong/non-monotonic plot points).
func TestSubscribeMultiSourceUsesLatestTimestamp(t *testing.T) {
	log := slog.New(slog.NewTextHandler(os.Stderr, nil))
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	now := time.Date(2026, 6, 19, 12, 0, 0, 0, time.UTC)
	// Fast source A with current timestamps.
	a := &seqSource{name: "A", seq: []datasource.Value{
		{Timestamp: now.Add(10 * time.Second), Data: 1.0, Quality: datasource.QualityGood},
		{Timestamp: now.Add(11 * time.Second), Data: 2.0, Quality: datasource.QualityGood},
		{Timestamp: now.Add(12 * time.Second), Data: 3.0, Quality: datasource.QualityGood},
		{Timestamp: now.Add(13 * time.Second), Data: 4.0, Quality: datasource.QualityGood},
	}}
	// Slow source B: a single sample with a much older timestamp.
	b := &seqSource{name: "B", seq: []datasource.Value{
		{Timestamp: now.Add(1 * time.Second), Data: 100.0, Quality: datasource.QualityGood},
	}}

	brk := broker.New(ctx, log)
	brk.Register(a)
	brk.Register(b)
	syn := New(t.TempDir(), brk, log)
	if err := syn.Connect(ctx); err != nil {
		t.Fatal(err)
	}
	if err := syn.AddSignal(SignalDef{
		Name: "diff",
		Graph: &Graph{Output: "out", Nodes: []GraphNode{
			{ID: "sa", Kind: "source", DS: "A", Signal: "x"},
			{ID: "sb", Kind: "source", DS: "B", Signal: "x"},
			{ID: "sub", Kind: "op", Op: "subtract", Inputs: []string{"sa", "sb"}},
			{ID: "out", Kind: "output", Inputs: []string{"sub"}},
		}},
	}); err != nil {
		t.Fatal(err)
	}

	ch := make(chan datasource.Value, 16)
	if _, err := syn.Subscribe(ctx, "diff", ch); err != nil {
		t.Fatal(err)
	}

	var last time.Time
	for i := 0; i < 4; i++ {
		select {
		case v := <-ch:
			// The stale source-B timestamp (t=1s) must never be used: every output
			// is stamped with the newest input time, so emits stay monotonic.
			if v.Timestamp.Equal(now.Add(1 * time.Second)) {
				t.Errorf("emit #%d used the stale source-B timestamp %s", i, v.Timestamp)
			}
			if !last.IsZero() && v.Timestamp.Before(last) {
				t.Errorf("emit #%d went backwards: %s before previous %s", i, v.Timestamp, last)
			}
			last = v.Timestamp
		case <-time.After(2 * time.Second):
			t.Fatalf("timeout waiting for emit #%d", i)
		}
	}
}