package chlog

import (
	"context"
	"time"
)

const (
	TailInterval = 2 * time.Second
	tailOverlap  = 5 * time.Second
)

// Tail streams the initial window then polls every interval. Each poll
// re-queries from a little before the last-seen timestamp and drops rows
// already emitted, so late-arriving rows inside the overlap still surface.
func Tail(ctx context.Context, c runner, f Filter, now func() time.Time, interval time.Duration, emit func(Row) error) error {
	if now == nil {
		now = time.Now
	}
	if interval <= 0 {
		interval = TailInterval
	}

	seen := map[uint64]time.Time{}
	lastTS := f.Since
	track := func(r Row) error {
		if ts := r.Time(); ts.After(lastTS) {
			lastTS = ts
		}
		seen[r.Key()] = r.Time()
		return emit(r)
	}

	first := f
	first.Until = now().UTC()
	first.Limit = 0
	if _, err := Page(ctx, c, first, 0, track); err != nil {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		return err
	}

	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}

		since := lastTS.Add(-tailOverlap)
		if since.Before(f.Since) {
			since = f.Since
		}
		pf := f
		pf.Since = since
		pf.Until = now().UTC()
		pf.Limit = 0
		if !pf.Since.Before(pf.Until) {
			continue
		}
		_, err := Page(ctx, c, pf, 0, func(r Row) error {
			if _, dup := seen[r.Key()]; dup {
				return nil
			}
			return track(r)
		})
		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			return err
		}

		floor := lastTS.Add(-2 * tailOverlap)
		for k, ts := range seen {
			if ts.Before(floor) {
				delete(seen, k)
			}
		}
	}
}