logstream/logsapi_test.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"math/rand"
	"net/http"
	"net/http/httptest"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"
)

// cursorVL plays VictoriaLogs for the two queries of /api/v1/logs: a time
// range sorted by _time, and an exact time. Like VictoriaLogs, it gives no
// order among rows of the same _time: they are shuffled on every query.
type cursorVL struct {
	mu      sync.Mutex
	rows    []map[string]any
	rnd     *rand.Rand
	queries []string
}

var cursorQuery = regexp.MustCompile(`^_time:\[([^,]+), ([^\]]+)] \((.*?)\)( \| sort by \(_time\))? \| limit (\d+)$`)

func (v *cursorVL) add(t time.Time, msg string) {
	v.mu.Lock()
	defer v.mu.Unlock()
	v.rows = append(v.rows, map[string]any{"_time": t.UTC().Format(time.RFC3339Nano), "_msg": msg, "host": "h1"})
}

func (v *cursorVL) RoundTrip(r *http.Request) (*http.Response, error) {
	_ = r.ParseForm()
	q := r.PostForm.Get("query")
	m := cursorQuery.FindStringSubmatch(q)
	if m == nil {
		return &http.Response{StatusCode: http.StatusBadRequest, Body: io.NopCloser(strings.NewReader("unexpected query " + q)), Header: http.Header{}}, nil
	}
	from, _ := time.Parse(time.RFC3339Nano, m[1])
	to, _ := time.Parse(time.RFC3339Nano, m[2])
	limit, _ := strconv.Atoi(m[5])

	v.mu.Lock()
	defer v.mu.Unlock()
	v.queries = append(v.queries, q)
	var out []map[string]any
	for _, row := range v.rows {
		t, _ := time.Parse(time.RFC3339Nano, row["_time"].(string))
		if !t.Before(from) && !t.After(to) {
			out = append(out, row)
		}
	}
	v.rnd.Shuffle(len(out), func(i, j int) { out[i], out[j] = out[j], out[i] })
	if m[4] != "" {
		sort.SliceStable(out, func(i, j int) bool {
			a, _ := time.Parse(time.RFC3339Nano, out[i]["_time"].(string))
			b, _ := time.Parse(time.RFC3339Nano, out[j]["_time"].(string))
			return a.Before(b)
		})
	}
	if len(out) > limit {
		out = out[:limit]
	}
	var b strings.Builder
	for _, row := range out {
		line, _ := json.Marshal(row)
		b.Write(line)
		b.WriteByte('\n')
	}
	return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(b.String())), Header: http.Header{}}, nil
}

func newCursorVL(seed int64) (*cursorVL, *Store) {
	vl := &cursorVL{rnd: rand.New(rand.NewSource(seed))}
	store := NewStore("http://vl", 10, 10, time.Second, nil)
	store.client.Transport = vl
	store.streamClient.Transport = vl
	return vl, store
}

// fill adds rows by groups of the same _time, some groups larger than a
// page, some with identical rows, and returns the multiset of the rows.
func fill(vl *cursorVL, start time.Time, groups []int, tag string) map[string]int {
	want := map[string]int{}
	t := start
	for g, size := range groups {
		for i := 0; i < size; i++ {
			msg := fmt.Sprintf("%s g%d r%d", tag, g, i%4) // r0..r3 repeat: identical rows
			vl.add(t, msg)
			want[rowKey(t, msg)]++
		}
		t = t.Add(time.Duration(1+g%3) * time.Microsecond)
	}
	return want
}

func rowKey(t time.Time, msg string) string { return t.UTC().Format(time.RFC3339Nano) + " " + msg }

func TestLogsCursorNoLossNoDuplicate(t *testing.T) {
	base := time.Date(2026, 11, 1, 10, 0, 0, 0, time.UTC)
	groups := []int{1, 3, 1, 1, 7, 2, 1, 12, 1, 4, 4, 1, 25, 1, 2}
	for _, limit := range []int{1, 2, 3, 5, 8, 1000} {
		for seed := int64(1); seed <= 5; seed++ {
			vl, store := newCursorVL(seed)
			want := fill(vl, base, groups, "a")
			got := map[string]int{}
			cur := logsCursor{T: base.Add(-time.Minute)}
			until := base.Add(time.Second)
			var last time.Time
			for calls := 0; ; calls++ {
				if calls > 500 {
					t.Fatalf("limit %d: no end", limit)
				}
				page, err := readLogsPage(context.Background(), store.Query, "*", cur, until, limit)
				if err != nil {
					t.Fatal(err)
				}
				if len(page.Rows) > limit {
					t.Fatalf("limit %d: page of %d rows", limit, len(page.Rows))
				}
				for _, row := range page.Rows {
					rt, _ := rowTime(row)
					if rt.Before(last) {
						t.Fatalf("limit %d: rows out of order (%s after %s)", limit, rt, last)
					}
					last = rt
					got[rowKey(rt, row["_msg"].(string))]++
				}
				var ok bool
				if cur, ok = decodeCursor(page.NextCursor); !ok {
					t.Fatalf("bad cursor %q", page.NextCursor)
				}
				if !page.HasMore {
					break
				}
				if len(page.Rows) == 0 {
					t.Fatalf("limit %d: has_more with an empty page", limit)
				}
			}
			if fmt.Sprint(got) != fmt.Sprint(want) {
				t.Fatalf("limit %d seed %d:\n got %v\nwant %v", limit, seed, got, want)
			}
		}
	}
}

// New logs arrive between the calls, as with a scheduled n8n workflow.
func TestLogsCursorFollowsNewLogs(t *testing.T) {
	vl, store := newCursorVL(7)
	base := time.Date(2026, 11, 1, 10, 0, 0, 0, time.UTC)
	want := map[string]int{}
	got := map[string]int{}
	cur := logsCursor{T: base}
	for round := 0; round < 6; round++ {
		start := base.Add(time.Duration(round) * time.Second)
		for k, n := range fill(vl, start, []int{2, 5, 1, 3}, fmt.Sprintf("round%d", round)) {
			want[k] += n
		}
		until := start.Add(500 * time.Millisecond) // after this round's rows
		for {
			page, err := readLogsPage(context.Background(), store.Query, "*", cur, until, 4)
			if err != nil {
				t.Fatal(err)
			}
			for _, row := range page.Rows {
				rt, _ := rowTime(row)
				got[rowKey(rt, row["_msg"].(string))]++
			}
			cur, _ = decodeCursor(page.NextCursor)
			if !page.HasMore {
				break
			}
		}
		// Polling again with nothing new returns nothing.
		page, err := readLogsPage(context.Background(), store.Query, "*", cur, until, 4)
		if err != nil || len(page.Rows) != 0 || page.HasMore {
			t.Fatalf("round %d: second poll returned %d rows (has_more %v, err %v)", round, len(page.Rows), page.HasMore, err)
		}
	}
	if fmt.Sprint(got) != fmt.Sprint(want) {
		t.Fatalf("got %v\nwant %v", got, want)
	}
}

func TestParseSince(t *testing.T) {
	now := time.Date(2026, 11, 1, 12, 0, 0, 0, time.UTC)
	c := logsCursor{T: now.Add(-time.Minute), K: 3, Q: "abcd"}
	for _, tc := range []struct {
		in   string
		want logsCursor
	}{
		{"", logsCursor{T: now.Add(-time.Hour)}},
		{"15m", logsCursor{T: now.Add(-15 * time.Minute)}},
		{"7d", logsCursor{T: now.Add(-7 * 24 * time.Hour)}},
		{"2026-11-01T08:30:00.5Z", logsCursor{T: time.Date(2026, 11, 1, 8, 30, 0, 5e8, time.UTC)}},
		{c.encode(), c},
	} {
		got, err := parseSince(tc.in, now)
		if err != nil || !got.T.Equal(tc.want.T) || got.K != tc.want.K || got.Q != tc.want.Q {
			t.Errorf("parseSince(%q) = %+v, %v; want %+v", tc.in, got, err, tc.want)
		}
	}
	for _, bad := range []string{"yesterday", "-5m", "e30"} {
		if _, err := parseSince(bad, now); err == nil {
			t.Errorf("parseSince(%q): expected an error", bad)
		}
	}
}

func TestLogsV1Handler(t *testing.T) {
	vl, store := newCursorVL(3)
	a := &API{store: store, rdns: NewReverseDNS(false, "")}
	old := time.Now().Add(-10 * time.Minute)
	fill(vl, old, []int{2, 1}, "x")
	vl.add(time.Now().Add(time.Hour), "in the future: after until")

	get := func(url string) (int, logsPage, map[string]string) {
		rec := httptest.NewRecorder()
		a.logsV1(rec, httptest.NewRequest("GET", url, nil))
		var page logsPage
		var e map[string]string
		_ = json.Unmarshal(rec.Body.Bytes(), &page)
		_ = json.Unmarshal(rec.Body.Bytes(), &e)
		return rec.Code, page, e
	}
	code, page, _ := get("/api/v1/logs?query=host:h1&since=1h")
	if code != http.StatusOK || len(page.Rows) != 3 || page.HasMore || page.NextCursor == "" {
		t.Fatalf("status %d, %d rows, has_more %v", code, len(page.Rows), page.HasMore)
	}
	if q := vl.queries[len(vl.queries)-1]; !strings.Contains(q, "(host:h1)") {
		t.Fatalf("query filter not applied: %s", q)
	}
	code, page, _ = get("/api/v1/logs?query=host:h1&since=" + page.NextCursor)
	if code != http.StatusOK || len(page.Rows) != 0 {
		t.Fatalf("second call: status %d, %d rows", code, len(page.Rows))
	}
	for url, wantCode := range map[string]string{
		"/api/v1/logs?query=" + "*%20|%20stats%20count()": "api_pipes",
		"/api/v1/logs?limit=0":                            "bad_limit",
		"/api/v1/logs?limit=20000":                        "bad_limit",
		"/api/v1/logs?lag=2h":                             "bad_lag",
		"/api/v1/logs?since=nope":                         "bad_since",
	} {
		if code, _, e := get(url); code != http.StatusBadRequest || e["code"] != wantCode {
			t.Errorf("%s: status %d, %v; want 400 %s", url, code, e, wantCode)
		}
	}
}

func TestLogsV1WaitsForSpool(t *testing.T) {
	spool, err := OpenSpool(t.TempDir(), 1<<20)
	if err != nil {
		t.Fatal(err)
	}
	if err := spool.Write([]byte(`{"_msg":"x"}`+"\n"), 1); err != nil {
		t.Fatal(err)
	}
	vl, _ := newCursorVL(1)
	store := NewStore("http://vl", 10, 10, time.Second, spool)
	store.streamClient.Transport = vl
	a := &API{store: store}
	rec := httptest.NewRecorder()
	a.logsV1(rec, httptest.NewRequest("GET", "/api/v1/logs?since=2026-11-01T10:00:00Z", nil))
	var page logsPage
	_ = json.Unmarshal(rec.Body.Bytes(), &page)
	c, _ := decodeCursor(page.NextCursor)
	if rec.Code != http.StatusOK || !page.Waiting || len(page.Rows) != 0 || !c.T.Equal(time.Date(2026, 11, 1, 10, 0, 0, 0, time.UTC)) || len(vl.queries) != 0 {
		t.Fatalf("status %d, waiting %v, %d rows, cursor %v, %d queries", rec.Code, page.Waiting, len(page.Rows), c.T, len(vl.queries))
	}
}