feat: add spkg.go for handling Substreams package resolution and loading

cmd/tools/substreams/daterange.go

@@ -0,0 +1,141 @@
+package substreams
+
+import (
+	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// ParseLogDateRange parses 1 or 2 args into a [start, end) time range.
+//
+// One arg supports:
+//   - relative duration ("1d", "2hr", "30m", "1 day ago")
+//   - "<start>/<end>" or "<start>:<end>" range (the colon split scans
+//     right-to-left so timezone-suffixed timestamps like "...10:00:00Z" work)
+//   - a single ISO timestamp, treated as [timestamp, now)
+//
+// Two args treats arg[0] as start and arg[1] as end.
+func ParseLogDateRange(args []string) (time.Time, time.Time, error) {
+	now := time.Now()
+	switch len(args) {
+	case 1:
+		return parseSingleLogDateArg(args[0], now)
+	case 2:
+		return parseTwoLogDateArgs(args[0], args[1], now)
+	default:
+		return time.Time{}, time.Time{}, fmt.Errorf("expected 1-2 date range args, got %d", len(args))
+	}
+}
+
+var relDurationRegex = regexp.MustCompile(`(?i)^\s*(\d+)\s*([a-z]+?)(?:\s+ago)?\s*$`)
+
+// parseRelativeDuration parses strings like "1d", "2hr", "30m", "1 day ago".
+func parseRelativeDuration(s string) (time.Duration, bool) {
+	m := relDurationRegex.FindStringSubmatch(strings.TrimSpace(s))
+	if m == nil {
+		return 0, false
+	}
+	n, err := strconv.ParseInt(m[1], 10, 64)
+	if err != nil || n <= 0 {
+		return 0, false
+	}
+	unit := strings.ToLower(m[2])
+	switch {
+	case strings.HasPrefix(unit, "d"):
+		return time.Duration(n) * 24 * time.Hour, true
+	case strings.HasPrefix(unit, "h"):
+		return time.Duration(n) * time.Hour, true
+	case strings.HasPrefix(unit, "m"):
+		return time.Duration(n) * time.Minute, true
+	}
+	return 0, false
+}
+
+func parseSingleLogDateArg(s string, now time.Time) (time.Time, time.Time, error) {
+	// Relative duration: "1d", "2hr", "30m", "1 day ago", …
+	if d, ok := parseRelativeDuration(s); ok {
+		return now.Add(-d), now, nil
+	}
+
+	// Range with "/" separator
+	if idx := strings.Index(s, "/"); idx >= 0 {
+		return parseLogRangeParts(s[:idx], s[idx+1:], now)
+	}
+
+	// Range with ":" separator — try each ":" from right to left so we
+	// find the inter-timestamp colon first (timestamps end with Z or ±hh:mm).
+	if start, end, ok := tryColonRangeSplit(s); ok {
+		return validateLogRange(start, end, now)
+	}
+
+	// Single ISO timestamp → [timestamp, now)
+	t, err := parseDateTime(s)
+	if err != nil {
+		return time.Time{}, time.Time{},
+			fmt.Errorf("unrecognized date-range format %q (try: 2h, 1 day ago, 2024-01-15T10:00:00Z, start:end)", s)
+	}
+	if t.After(now) {
+		return time.Time{}, time.Time{},
+			fmt.Errorf("timestamp %q is in the future — provide a past timestamp as start of range", s)
+	}
+	return t, now, nil
+}
+
+func parseTwoLogDateArgs(s1, s2 string, now time.Time) (time.Time, time.Time, error) {
+	start, err := parseDateTime(strings.TrimSpace(s1))
+	if err != nil {
+		return time.Time{}, time.Time{}, fmt.Errorf("parsing start %q: %w", s1, err)
+	}
+	end, err := parseDateTime(strings.TrimSpace(s2))
+	if err != nil {
+		return time.Time{}, time.Time{}, fmt.Errorf("parsing end %q: %w", s2, err)
+	}
+	return validateLogRange(start, end, now)
+}
+
+func parseLogRangeParts(s1, s2 string, now time.Time) (time.Time, time.Time, error) {
+	start, err := parseDateTime(strings.TrimSpace(s1))
+	if err != nil {
+		return time.Time{}, time.Time{}, fmt.Errorf("parsing start %q: %w", s1, err)
+	}
+	end, err := parseDateTime(strings.TrimSpace(s2))
+	if err != nil {
+		return time.Time{}, time.Time{}, fmt.Errorf("parsing end %q: %w", s2, err)
+	}
+	return validateLogRange(start, end, now)
+}
+
+// tryColonRangeSplit attempts to split s at a colon that separates two valid
+// timestamps, scanning from right to left so we find inter-timestamp colons first.
+func tryColonRangeSplit(s string) (time.Time, time.Time, bool) {
+	for i := len(s) - 1; i >= 0; i-- {
+		if s[i] != ':' {
+			continue
+		}
+		left := strings.TrimSpace(s[:i])
+		right := strings.TrimSpace(s[i+1:])
+		if left == "" || right == "" {
+			continue
+		}
+		start, err1 := parseDateTime(left)
+		end, err2 := parseDateTime(right)
+		if err1 == nil && err2 == nil {
+			return start, end, true
+		}
+	}
+	return time.Time{}, time.Time{}, false
+}
+
+func validateLogRange(start, end, now time.Time) (time.Time, time.Time, error) {
+	if start.After(now) && end.After(now) {
+		return time.Time{}, time.Time{}, fmt.Errorf("both start and end are in the future")
+	}
+	if !end.After(start) {
+		return time.Time{}, time.Time{},
+			fmt.Errorf("end time %s is not after start time %s",
+				end.UTC().Format(time.RFC3339), start.UTC().Format(time.RFC3339))
+	}
+	return start, end, nil
+}

cmd/tools/substreams/daterange_test.go

@@ -0,0 +1,98 @@
+package substreams
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseLogDateRange(t *testing.T) {
+	now := time.Now()
+
+	t.Run("relative duration single arg", func(t *testing.T) {
+		start, end, err := ParseLogDateRange([]string{"2h"})
+		require.NoError(t, err)
+		assert.WithinDuration(t, now, end, 2*time.Second)
+		assert.WithinDuration(t, now.Add(-2*time.Hour), start, 2*time.Second)
+	})
+
+	t.Run("relative duration with 'ago' suffix", func(t *testing.T) {
+		start, end, err := ParseLogDateRange([]string{"1 day ago"})
+		require.NoError(t, err)
+		assert.WithinDuration(t, now, end, 2*time.Second)
+		assert.WithinDuration(t, now.Add(-24*time.Hour), start, 2*time.Second)
+	})
+
+	t.Run("slash-separated range", func(t *testing.T) {
+		start, end, err := ParseLogDateRange([]string{"2024-01-15T10:00:00Z/2024-01-15T12:00:00Z"})
+		require.NoError(t, err)
+		expectedStart, _ := time.Parse(time.RFC3339, "2024-01-15T10:00:00Z")
+		expectedEnd, _ := time.Parse(time.RFC3339, "2024-01-15T12:00:00Z")
+		assert.Equal(t, expectedStart, start)
+		assert.Equal(t, expectedEnd, end)
+	})
+
+	t.Run("colon-separated range", func(t *testing.T) {
+		start, end, err := ParseLogDateRange([]string{"2024-01-15T10:00:00Z:2024-01-15T12:00:00Z"})
+		require.NoError(t, err)
+		expectedStart, _ := time.Parse(time.RFC3339, "2024-01-15T10:00:00Z")
+		expectedEnd, _ := time.Parse(time.RFC3339, "2024-01-15T12:00:00Z")
+		assert.Equal(t, expectedStart, start)
+		assert.Equal(t, expectedEnd, end)
+	})
+
+	t.Run("two args", func(t *testing.T) {
+		start, end, err := ParseLogDateRange([]string{"2024-01-15T10:00:00Z", "2024-01-15T12:00:00Z"})
+		require.NoError(t, err)
+		expectedStart, _ := time.Parse(time.RFC3339, "2024-01-15T10:00:00Z")
+		expectedEnd, _ := time.Parse(time.RFC3339, "2024-01-15T12:00:00Z")
+		assert.Equal(t, expectedStart, start)
+		assert.Equal(t, expectedEnd, end)
+	})
+
+	t.Run("end before start fails", func(t *testing.T) {
+		_, _, err := ParseLogDateRange([]string{"2024-01-15T12:00:00Z/2024-01-15T10:00:00Z"})
+		require.Error(t, err)
+	})
+
+	t.Run("zero args fails", func(t *testing.T) {
+		_, _, err := ParseLogDateRange(nil)
+		require.Error(t, err)
+	})
+
+	t.Run("future single timestamp fails", func(t *testing.T) {
+		future := now.Add(24 * time.Hour).Format(time.RFC3339)
+		_, _, err := ParseLogDateRange([]string{future})
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "future")
+	})
+}
+
+func TestParseRelativeDuration(t *testing.T) {
+	tests := []struct {
+		input string
+		want  time.Duration
+		ok    bool
+	}{
+		{"1d", 24 * time.Hour, true},
+		{"2hr", 2 * time.Hour, true},
+		{"30m", 30 * time.Minute, true},
+		{"1 day ago", 24 * time.Hour, true},
+		{"2 hours ago", 2 * time.Hour, true},
+		{"0d", 0, false}, // zero is rejected
+		{"abc", 0, false},
+		{"", 0, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.input, func(t *testing.T) {
+			got, ok := parseRelativeDuration(tt.input)
+			assert.Equal(t, tt.ok, ok)
+			if tt.ok {
+				assert.Equal(t, tt.want, got)
+			}
+		})
+	}
+}

cmd/tools/substreams/endpoint.go

@@ -0,0 +1,55 @@
+package substreams
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/streamingfast/firehose-core/cmd/tools/stylex"
+	networks "github.com/streamingfast/firehose-networks"
+)
+
+// ResolveEndpoint returns the endpoint to use: the explicit flag value if set,
+// otherwise inferred from the k8s namespace via the networks registry. When
+// inferred, a one-line note is printed to stdout for the user.
+func ResolveEndpoint(flagValue, namespace string) (string, error) {
+	if flagValue != "" {
+		return flagValue, nil
+	}
+	if namespace == "" {
+		return "", fmt.Errorf("could not infer endpoint: no namespace found in log entry; provide --endpoint explicitly")
+	}
+
+	endpoint := InferEndpointFromNamespace(namespace)
+	if endpoint == "" {
+		return "", fmt.Errorf(
+			"could not infer Substreams endpoint from namespace %q; provide --endpoint explicitly (e.g. --endpoint mainnet.eth.streamingfast.io:443)",
+			namespace,
+		)
+	}
+
+	fmt.Printf("%s %s %s\n",
+		stylex.Label("Endpoint (inferred):"),
+		stylex.Value(endpoint),
+		stylex.Dimf("(from namespace %q)", namespace),
+	)
+	return endpoint, nil
+}
+
+// InferEndpointFromNamespace tries to find a Substreams endpoint for the given k8s namespace.
+// Namespaces are typically "<chain>-<network>" (e.g. "eth-polygon-mainnet").
+// It progressively strips leading dash-separated segments until a match is found:
+//
+//	eth-polygon-mainnet → polygon-mainnet → mainnet
+func InferEndpointFromNamespace(namespace string) string {
+	for s := namespace; s != ""; {
+		if ep := networks.GetSubstreamsEndpoint(s); ep != "" {
+			return ep
+		}
+		_, rest, found := strings.Cut(s, "-")
+		if !found {
+			break
+		}
+		s = rest
+	}
+	return ""
+}

cmd/tools/substreams/logs.go

@@ -23,6 +23,7 @@ func NewToolsLogsCmd(logger *zap.Logger) *cobra.Command {
 Note: Currently only GCP Cloud Logging backend is supported.`,
 	}
 
+	cmd.AddCommand(NewToolsLogsConnectionCmd(logger))
 	cmd.AddCommand(NewToolsLogsConnectionsCmd(logger))
 	cmd.AddCommand(NewToolsLogsReexecCmd(logger))
 

cmd/tools/substreams/logs/backend.go

@@ -10,13 +10,17 @@ type LogEntry struct {
 	// Common fields from jsonPayload
 	Message          string
 	TraceID          string
+	SessionID        string
 	UserID           string
 	IPAddress        string
 	OutputModule     string
 	OutputModuleHash string
 	StartBlock       int64
 	StopBlock        uint64
+	Cursor           string
 	ProductionMode   bool
+	FinalBlocksOnly  bool
+	NoopMode         bool
 	Timestamp        string
 
 	// Stats-specific fields (only present for "substreams request stats")