feat: add opal subcommands for pipeline linting and reference

cmd_opal.go

@@ -0,0 +1,185 @@
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/spf13/pflag"
+)
+
+var (
+	flagsOpal       *pflag.FlagSet
+	flagOpalFile    string
+	flagOpalDataset string
+)
+
+var ErrOpalUsage = ObserveError{Msg: "usage: observe opal <check|verbs|functions|validate-ingest> [args...]"}
+
+func init() {
+	flagsOpal = pflag.NewFlagSet("opal", pflag.ContinueOnError)
+	flagsOpal.StringVarP(&flagOpalFile, "file", "f", "", "Read OPAL pipeline from file instead of argument")
+	flagsOpal.StringVar(&flagOpalDataset, "dataset", "", "Source dataset ID for validate-ingest subcommand")
+	RegisterCommand(&Command{
+		Name:  "opal",
+		Help:  "Validate and inspect OPAL pipelines and functions.",
+		Flags: flagsOpal,
+		Func:  cmdOpal,
+	})
+}
+
+func cmdOpal(fa FuncArgs) error {
+	if len(fa.args) < 2 {
+		return ErrOpalUsage
+	}
+	switch fa.args[1] {
+	case "check":
+		return cmdOpalCheck(fa)
+	case "verbs":
+		return cmdOpalVerbs(fa)
+	case "functions":
+		return cmdOpalFunctions(fa)
+	case "validate-ingest":
+		return cmdOpalValidateIngest(fa)
+	default:
+		return ObserveError{Msg: fmt.Sprintf("unknown opal subcommand %q; expected check, verbs, functions, or validate-ingest", fa.args[1])}
+	}
+}
+
+// cmdOpalVerbs and cmdOpalFunctions are implemented in ot_opal.go (issue #6).
+// cmdOpalValidateIngest is implemented in cmd_opal_validate.go (issue #7).
+
+// gqlCheckQueries validates an OPAL pipeline using the checkQueries GraphQL operation.
+//
+// Actual API schema (discovered via integration tests):
+//   - Input: MultiStageQueryInput { outputStage: String!, stages: [StageQueryInput!]! }
+//   - StageQueryInput: { stageID: String!, pipeline: String!, input: [InputDefinitionInput!]! }
+//   - Returns: [CompilationResult!]  (array, one per stage)
+//   - CompilationResult.parsedPipeline.errors: [PipelineSymbol!] { col, row, text, type }
+//   - CompilationResult.parsedPipeline.warnings: [PipelineWarning!] { kind, symbol { col, row, text } }
+//   - CompilationResult.resultSchema.fieldList: [{ name }]
+//   - Errors with text=="" mean "compilation requires an input dataset" (not a real syntax error)
+var gqlCheckQueries = compileGqlQuery(
+	`query CheckQueries($queries: MultiStageQueryInput!) {
+		checkQueries(queries: $queries) {
+			parsedPipeline {
+				errors { col row text }
+				warnings { kind symbol { col row } }
+			}
+			resultSchema { fieldList { name } }
+		}
+	}`,
+	"data", "checkQueries", "0",
+)
+
+// opalPos holds row:col position info from the API response.
+type opalPos struct {
+	row string
+	col string
+}
+
+func extractPos(sym any) opalPos {
+	p := opalPos{}
+	if m, ok := sym.(object); ok {
+		if v, ok := m["row"].(string); ok {
+			p.row = v
+		}
+		if v, ok := m["col"].(string); ok {
+			p.col = v
+		}
+	}
+	return p
+}
+
+func cmdOpalCheck(fa FuncArgs) error {
+	var pipeline string
+
+	if flagOpalFile != "" {
+		data, err := os.ReadFile(flagOpalFile)
+		if err != nil {
+			return fmt.Errorf("opal check: could not read file %q: %w", flagOpalFile, err)
+		}
+		pipeline = string(data)
+	} else if len(fa.args) >= 3 {
+		pipeline = fa.args[2]
+	} else {
+		return ObserveError{Msg: "usage: observe opal check <pipeline> | observe opal check --file <path>"}
+	}
+
+	stage := object{
+		"stageID":  "stage-1",
+		"pipeline": pipeline,
+		"input":    array{},
+	}
+	queries := object{
+		"outputStage": "stage-1",
+		"stages":      array{stage},
+	}
+
+	result, err := gqlCheckQueries.query(fa.cfg, fa.op, fa.hc, object{"queries": queries})
+	if err != nil {
+		return err
+	}
+
+	res, ok := result.(object)
+	if !ok {
+		return fmt.Errorf("opal check: unexpected response type")
+	}
+
+	// Extract parsedPipeline
+	pp, _ := res["parsedPipeline"].(object)
+	var errors []object
+	var warnings []object
+
+	if pp != nil {
+		if errList, ok := pp["errors"].(array); ok {
+			for _, e := range errList {
+				if m, ok := e.(object); ok {
+					// Skip errors with empty text — these indicate "compilation requires an input
+					// dataset" and are not real syntax errors in the pipeline itself.
+					if text, _ := m["text"].(string); text != "" {
+						errors = append(errors, m)
+					}
+				}
+			}
+		}
+		if warnList, ok := pp["warnings"].(array); ok {
+			for _, w := range warnList {
+				if m, ok := w.(object); ok {
+					warnings = append(warnings, m)
+				}
+			}
+		}
+	}
+
+	if len(errors) > 0 {
+		for _, e := range errors {
+			text, _ := e["text"].(string)
+			row, _ := e["row"].(string)
+			col, _ := e["col"].(string)
+			fmt.Fprintf(fa.op, "ERROR %s:%s: %s\n", row, col, text)
+		}
+		return ObserveError{Msg: "opal check: pipeline has errors"}
+	}
+
+	for _, w := range warnings {
+		kind, _ := w["kind"].(string)
+		sym, _ := w["symbol"].(object)
+		pos := extractPos(sym)
+		fmt.Fprintf(fa.op, "WARN %s %s:%s\n", kind, pos.row, pos.col)
+	}
+
+	// Print OK and optionally the result schema fields
+	fmt.Fprintf(fa.op, "OK\n")
+	if schema, ok := res["resultSchema"].(object); ok {
+		if fields, ok := schema["fieldList"].(array); ok {
+			for _, f := range fields {
+				if fm, ok := f.(object); ok {
+					name, _ := fm["name"].(string)
+					fmt.Fprintf(fa.op, "  %s\n", name)
+				}
+			}
+		}
+	}
+
+	return nil
+}

cmd_opal_extra_test.go

@@ -0,0 +1,142 @@
+package main
+
+import (
+	"net/http"
+	"strings"
+	"testing"
+)
+
+// validateIngestResponseHelper builds a mock GraphQL response for validateIngestFilterExpression.
+func validateIngestResponseHelper(diagsJSON string) string {
+	return `{"data":{"validateIngestFilterExpression":` + diagsJSON + `}}`
+}
+
+// TestCmdOpalCheckEmptyPipeline verifies that an empty pipeline string is sent to
+// the API rather than triggering a local usage error.
+func TestCmdOpalCheckEmptyPipeline(t *testing.T) {
+	resp := checkQueriesResponse("[]", "[]", `[]`)
+	fix := startFixture(t,
+		testRequest{"/v1/meta", 200, resp},
+	)
+	RunCommandWithConfig(fix.cfg, fix.fs, fix.op, []string{"opal", "check", ""}, fix.hc)
+	fix.Assert()
+	out := fix.op.OutputBuf.String()
+	if !strings.Contains(out, "OK") {
+		t.Errorf("expected OK for empty pipeline (accepted by API), got: %q", out)
+	}
+}
+
+// TestCmdOpalCheckMultipleErrors verifies that all errors are printed, not just the first.
+func TestCmdOpalCheckMultipleErrors(t *testing.T) {
+	resp := checkQueriesResponse(
+		`[{"col":"1","row":"1","text":"bad_verb"},{"col":"10","row":"1","text":"bad_arg"}]`,
+		"[]",
+		"null",
+	)
+	fix := startFixture(t,
+		testRequest{"/v1/meta", 200, resp},
+	)
+	mustPanic(t, func() {
+		RunCommandWithConfig(fix.cfg, fix.fs, fix.op, []string{"opal", "check", "bad stuff"}, fix.hc)
+	})
+	fix.Assert()
+	out := fix.op.OutputBuf.String()
+	if !strings.Contains(out, "bad_verb") {
+		t.Errorf("expected first error text in output, got: %q", out)
+	}
+	if !strings.Contains(out, "bad_arg") {
+		t.Errorf("expected second error text in output, got: %q", out)
+	}
+}
+
+// TestCmdOpalCheckNetworkError verifies that a network-level failure is surfaced as an error.
+func TestCmdOpalCheckNetworkError(t *testing.T) {
+	// Use a server that immediately returns 500.
+	fix := startFixture(t,
+		testRequest{"/v1/meta", 500, `{"errors":[{"message":"internal server error"}]}`},
+	)
+	mustPanic(t, func() {
+		RunCommandWithConfig(fix.cfg, fix.fs, fix.op, []string{"opal", "check", "filter true"}, fix.hc)
+	})
+	fix.Assert()
+}
+
+// errorHttpClient always returns an error (simulates connection failure).
+type errorHttpClient struct{}
+
+func (e *errorHttpClient) Do(req *http.Request) (*http.Response, error) {
+	return nil, &httpDialError{"connection refused"}
+}
+
+type httpDialError struct{ msg string }
+
+func (e *httpDialError) Error() string { return e.msg }
+
+// TestCmdOpalCheckConnectionError verifies behavior when HTTP connection fails.
+func TestCmdOpalCheckConnectionError(t *testing.T) {
+	fix := startFixture(t) // no requests expected
+	mustPanic(t, func() {
+		RunCommandWithConfig(fix.cfg, fix.fs, fix.op, []string{"opal", "check", "filter true"}, &errorHttpClient{})
+	})
+	// The error should surface, not panic internally
+}
+
+// TestCmdOpalVerbsNetworkError verifies that a network failure from opal verbs is surfaced.
+func TestCmdOpalVerbsNetworkError(t *testing.T) {
+	fix := startFixture(t,
+		testRequest{"/v1/meta", 500, `{"errors":[{"message":"internal server error"}]}`},
+	)
+	mustPanic(t, func() {
+		RunCommandWithConfig(fix.cfg, fix.fs, fix.op, []string{"opal", "verbs"}, fix.hc)
+	})
+	fix.Assert()
+}
+
+// TestCmdOpalFunctionsNetworkError verifies that a network failure from opal functions is surfaced.
+func TestCmdOpalFunctionsNetworkError(t *testing.T) {
+	fix := startFixture(t,
+		testRequest{"/v1/meta", 500, `{"errors":[{"message":"internal server error"}]}`},
+	)
+	mustPanic(t, func() {
+		RunCommandWithConfig(fix.cfg, fix.fs, fix.op, []string{"opal", "functions"}, fix.hc)
+	})
+	fix.Assert()
+}
+
+// TestCmdOpalValidateIngestSuccessEdge verifies successful ingest filter validation.
+func TestCmdOpalValidateIngestSuccessEdge(t *testing.T) {
+	resp := validateIngestResponseHelper(`[]`)
+	fix := startFixture(t,
+		testRequest{"/v1/meta", 200, resp},
+	)
+	flagOpalDataset = ""
+	RunCommandWithConfig(fix.cfg, fix.fs, fix.op, []string{"opal", "validate-ingest", "--dataset", "42918275", "filter true"}, fix.hc)
+	flagOpalDataset = ""
+	fix.Assert()
+	out := fix.op.OutputBuf.String()
+	if !strings.Contains(out, "OK") {
+		t.Errorf("expected OK in output, got: %q", out)
+	}
+}
+
+// TestCmdOpalValidateIngestWarnings verifies that messages from validate-ingest are treated as errors.
+// (The API only returns errors, not warnings, for ingest filter validation.)
+func TestCmdOpalValidateIngestWarnings(t *testing.T) {
+	resp := validateIngestResponseHelper(
+		`[{"message":"some validation message"}]`,
+	)
+	fix := startFixture(t,
+		testRequest{"/v1/meta", 200, resp},
+	)
+	flagOpalDataset = ""
+	mustPanic(t, func() {
+		RunCommandWithConfig(fix.cfg, fix.fs, fix.op, []string{"opal", "validate-ingest", "--dataset", "42918275", "filter true"}, fix.hc)
+	})
+	flagOpalDataset = ""
+	fix.Assert()
+	out := fix.op.OutputBuf.String()
+	// All messages from validateIngestFilterExpression are treated as errors
+	if !strings.Contains(out, "ERROR") {
+		t.Errorf("expected ERROR for any validateIngest message, got: %q", out)
+	}
+}