diff --git a/platform/README.md b/platform/README.md index 9fdcdd0ca..e34b69eb8 100644 --- a/platform/README.md +++ b/platform/README.md @@ -13,6 +13,7 @@ Cross-domain packages shared by SubmitQueue, Stovepipe, and other services in th - **lifecycle/** — `Component` start/stop interface and ordered `Group` composition for runnable subsystems. - **pipeline/** — Typed engine that assembles a service's queue topology into one lifecycle component. - **publish/** — Topic-key publish plumbing and the message-ID convention that controls deduplication. +- **conduit/** — Phabricator Conduit client: one generic `Call` that encodes JSON parameters with the API token, and decodes the result envelope into a typed `*Error` or the caller's result. Typed methods stay with each caller. - **buildkite/**, **githubactions/** — Vendor CI clients over `platform/http`: the REST calls and provider-specific vocabulary (state strings, id encoding) shared by every domain's `BuildRunner` backend. They deliberately define no interface, so they are plumbing rather than extensions — each domain keeps its own `BuildRunner` contract under `{domain}/extension/buildrunner` and adapts these clients to it. - **base/** — Shared domain entities (`change`, `messagequeue`, and related subpackages). Root package `base` is documentation-only. - **extension/** — Shared extension interfaces and implementations reused across domains (`counter`, `messagequeue`, and backends such as `mysql`). A package belongs here only if it defines a behavioral interface with its `Config` and `Factory` interface; a vendor client with no interface belongs directly under `platform/`. diff --git a/platform/conduit/BUILD.bazel b/platform/conduit/BUILD.bazel new file mode 100644 index 000000000..5e3df0855 --- /dev/null +++ b/platform/conduit/BUILD.bazel @@ -0,0 +1,20 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["conduit.go"], + importpath = "github.com/uber/submitqueue/platform/conduit", + visibility = ["//visibility:public"], + deps = ["//platform/http:go_default_library"], +) + +go_test( + name = "go_default_test", + srcs = ["conduit_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/http:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/platform/conduit/README.md b/platform/conduit/README.md new file mode 100644 index 000000000..2ce1aefc6 --- /dev/null +++ b/platform/conduit/README.md @@ -0,0 +1,47 @@ +# Conduit client + +Conduit is Phabricator's HTTP API. This package is the one way SubmitQueue code calls it: it encodes a method's parameters, authenticates the call, and turns Conduit's reply into either the caller's typed result or a typed error. It holds a single generic `Call` and no per-method wrappers. + +## Setting it up + +Build the client once in wiring and share it; it is safe for concurrent use. + +- Start from an `*http.Client` whose transport resolves relative paths against the Phabricator host (`platform/http.NewClient(endpoint)` installs a `BaseURLTransport`). That client owns the timeout and connection limits. +- Wrap it with `NewClient(httpClient, token)`, where `token` is a Conduit API token. One client serves one host and one token. + +The package never reads configuration or credentials itself. + +## What a call sends + +`Call(ctx, method, params, result)` posts a form to `/api/` with two fields: + +- `params`: the caller's parameters as a JSON object, with the token added under `__conduit__`. Parameters may be a map or a struct, or nil for none; anything that does not encode as a JSON object is rejected before a request is sent. +- `output`: `json`. + +The context bounds the request, so cancelling it or hitting its deadline aborts the call. + +### Why the token is on the client, not in a transport + +Other HTTP clients in `platform/` leave auth to a transport the wiring installs. Conduit cannot work that way: once the parameters are JSON, it reads the token only from `__conduit__` and silently ignores an `api.token` query or form field, answering `ERR-INVALID-SESSION`. A transport would have to decode and rewrite the request body, so the token is passed to `NewClient` instead. + +## What a call returns + +| Conduit's reply | `Call` returns | +|---|---| +| HTTP 200, `result` set | `nil`, with `result` decoded into the caller's value | +| HTTP 200, `result` is `null` or `[]` | `nil`, with the caller's value untouched. PHP encodes an empty map as `[]`, so an empty array means "nothing", not "wrong shape". | +| HTTP 200, `error_code` set | an `*Error` carrying `Code` and `Info` | +| any other HTTP status | a `platform/http.StatusError` carrying the status and body | +| a body that is not JSON, or a result that does not fit the caller's type | a plain error | + +Every error is wrapped with the method name. + +### Classifying failures + +- A `StatusError` is classified by `platform/errs/http` like any other HTTP failure: a 5xx or 429 is retryable, other statuses are not. +- An `*Error` is not an HTTP status, because Conduit reports its own failures with HTTP 200. That includes a missing or rejected token (`ERR-INVALID-SESSION`). With no classifier for it, such an error is non-retryable infra by default, so a bad token dead-letters rather than retrying forever. A caller that knows a code means something else, such as "not applicable", matches on `Code` itself. + +## What is deliberately not here + +- **Per-method wrappers.** Each caller keeps its own typed request and result for the methods it calls, and decides what a missing object means for it. A method only one caller uses never lands in this package. +- **Retries, logging and metrics.** Retry decisions belong to the consumer's error classifiers, and callers already log and record metrics in their own scope. diff --git a/platform/conduit/conduit.go b/platform/conduit/conduit.go new file mode 100644 index 000000000..2265c7b5f --- /dev/null +++ b/platform/conduit/conduit.go @@ -0,0 +1,145 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package conduit calls Phabricator's Conduit API: it encodes a method's +// parameters, authenticates the call and decodes Conduit's result envelope. +// +// It holds one generic Call and no per-method wrappers. Each caller keeps its +// own typed requests and results and decides what a missing object means for +// it, so a method only one caller uses never lands here. +// +// The API token travels inside the JSON parameters, as "__conduit__.token". +// Conduit ignores an "api.token" field once the parameters are JSON, so auth +// cannot be left to a transport the way it is for other HTTP APIs. +package conduit + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + + phttp "github.com/uber/submitqueue/platform/http" +) + +// Error is a failure Conduit reports in its result envelope. Conduit answers +// such failures, including a missing or invalid token (ERR-INVALID-SESSION), +// with HTTP 200, so they never surface as a platform/http.StatusError. +type Error struct { + // Code is Conduit's error_code, e.g. ERR-INVALID-SESSION. + Code string + // Info is Conduit's error_info, a human-readable description. + Info string +} + +func (e *Error) Error() string { + return fmt.Sprintf("conduit error %s: %s", e.Code, e.Info) +} + +// Client calls Conduit methods with one API token. It is safe for concurrent +// use. +type Client struct { + httpClient *http.Client + token string +} + +// NewClient wraps httpClient, which must resolve relative paths against the +// Phabricator host (e.g. via platform/http.BaseURLTransport) and carries the +// caller's timeout and connection limits. token authenticates every call. +func NewClient(httpClient *http.Client, token string) *Client { + return &Client{httpClient: httpClient, token: token} +} + +type envelope struct { + Result json.RawMessage `json:"result"` + ErrorCode *string `json:"error_code"` + ErrorInfo *string `json:"error_info"` +} + +// Call invokes method with params, which must encode as a JSON object (nil +// sends none), and decodes the result into result unless it is nil. A non-200 +// reply is a *platform/http.StatusError and a Conduit error_code is an *Error. +// A null result, or an empty array (PHP's encoding of an empty map), leaves +// result untouched. +func (c *Client) Call(ctx context.Context, method string, params, result any) error { + body, err := c.encodeParams(params) + if err != nil { + return fmt.Errorf("conduit %s: %w", method, err) + } + + status, respBody, err := phttp.SendRequest(ctx, c.httpClient, http.MethodPost, "/api/"+url.PathEscape(method), body, setFormHeaders) + if err != nil { + return fmt.Errorf("conduit %s: %w", method, err) + } + if status != http.StatusOK { + return fmt.Errorf("conduit %s: %w", method, phttp.NewStatusError(status, respBody)) + } + + var env envelope + if err := json.Unmarshal(respBody, &env); err != nil { + return fmt.Errorf("conduit %s: decode response: %w", method, err) + } + if env.ErrorCode != nil && *env.ErrorCode != "" { + info := "" + if env.ErrorInfo != nil { + info = *env.ErrorInfo + } + return fmt.Errorf("conduit %s: %w", method, &Error{Code: *env.ErrorCode, Info: info}) + } + if result == nil || isEmpty(env.Result) { + return nil + } + if err := json.Unmarshal(env.Result, result); err != nil { + return fmt.Errorf("conduit %s: decode result: %w", method, err) + } + return nil +} + +func (c *Client) encodeParams(params any) ([]byte, error) { + fields := map[string]json.RawMessage{} + if params != nil { + raw, err := json.Marshal(params) + if err != nil { + return nil, fmt.Errorf("marshal params: %w", err) + } + if err := json.Unmarshal(raw, &fields); err != nil { + return nil, fmt.Errorf("params must encode as a JSON object: %w", err) + } + if fields == nil { + fields = map[string]json.RawMessage{} + } + } + auth, err := json.Marshal(map[string]string{"token": c.token}) + if err != nil { + return nil, fmt.Errorf("marshal auth: %w", err) + } + fields["__conduit__"] = auth + + encoded, err := json.Marshal(fields) + if err != nil { + return nil, fmt.Errorf("marshal params: %w", err) + } + return []byte(url.Values{"params": {string(encoded)}, "output": {"json"}}.Encode()), nil +} + +func isEmpty(raw json.RawMessage) bool { + trimmed := bytes.TrimSpace(raw) + return len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) || bytes.Equal(trimmed, []byte("[]")) +} + +func setFormHeaders(req *http.Request) { + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") +} diff --git a/platform/conduit/conduit_test.go b/platform/conduit/conduit_test.go new file mode 100644 index 000000000..976c50940 --- /dev/null +++ b/platform/conduit/conduit_test.go @@ -0,0 +1,258 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package conduit + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + phttp "github.com/uber/submitqueue/platform/http" +) + +const testToken = "api-test-token" + +// receivedCall is what the fake Conduit server saw. +type receivedCall struct { + method string + path string + contentType string + output string + params map[string]any +} + +// fakeConduit records every call it receives and answers each with one reply. +type fakeConduit struct { + mu sync.Mutex + calls []receivedCall +} + +func (f *fakeConduit) received() []receivedCall { + f.mu.Lock() + defer f.mu.Unlock() + return append([]receivedCall(nil), f.calls...) +} + +func newTestClient(t *testing.T, status int, body string) (*Client, *fakeConduit) { + t.Helper() + fake := &fakeConduit{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !assert.NoError(t, r.ParseForm()) { + return + } + var params map[string]any + _ = json.Unmarshal([]byte(r.PostForm.Get("params")), ¶ms) + fake.mu.Lock() + fake.calls = append(fake.calls, receivedCall{ + method: r.Method, + path: r.URL.Path, + contentType: r.Header.Get("Content-Type"), + output: r.PostForm.Get("output"), + params: params, + }) + fake.mu.Unlock() + w.WriteHeader(status) + _, _ = io.WriteString(w, body) + })) + t.Cleanup(server.Close) + httpClient, err := phttp.NewClient(server.URL) + require.NoError(t, err) + return NewClient(httpClient, testToken), fake +} + +func TestClient_Call_SendsParamsWithToken(t *testing.T) { + tests := []struct { + name string + params any + wantParams map[string]any + }{ + { + name: "map params", + params: map[string]any{"ids": []int{72558007}}, + wantParams: map[string]any{ + "ids": []any{float64(72558007)}, + "__conduit__": map[string]any{"token": testToken}, + }, + }, + { + name: "struct params", + params: struct { + Constraints map[string][]int `json:"constraints"` + }{Constraints: map[string][]int{"ids": {30026137}}}, + wantParams: map[string]any{ + "constraints": map[string]any{"ids": []any{float64(30026137)}}, + "__conduit__": map[string]any{"token": testToken}, + }, + }, + { + name: "nil params send only the token", + wantParams: map[string]any{"__conduit__": map[string]any{"token": testToken}}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client, fake := newTestClient(t, http.StatusOK, `{"result": {}, "error_code": null, "error_info": null}`) + + require.NoError(t, client.Call(context.Background(), "differential.querydiffs", tt.params, nil)) + + calls := fake.received() + require.Len(t, calls, 1) + call := calls[0] + assert.Equal(t, http.MethodPost, call.method) + assert.Equal(t, "/api/differential.querydiffs", call.path) + assert.Equal(t, "application/x-www-form-urlencoded", call.contentType) + assert.Equal(t, "json", call.output) + assert.Equal(t, tt.wantParams, call.params) + }) + } +} + +func TestClient_Call_RejectsNonObjectParams(t *testing.T) { + tests := []struct { + name string + params any + }{ + {name: "array", params: []int{1, 2}}, + {name: "string", params: "D123"}, + {name: "number", params: 7}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client, fake := newTestClient(t, http.StatusOK, `{"result": {}}`) + + require.Error(t, client.Call(context.Background(), "differential.querydiffs", tt.params, nil)) + assert.Empty(t, fake.received(), "a request with malformed params must not be sent") + }) + } +} + +func TestClient_Call_DecodesResponse(t *testing.T) { + type diff struct { + ID string `json:"id"` + } + + tests := []struct { + name string + status int + body string + want map[string]diff + wantErr bool + wantStatus int + wantErrorCode string + }{ + { + name: "object result", + status: http.StatusOK, + body: `{"result": {"72558007": {"id": "72558007"}}, "error_code": null, "error_info": null}`, + want: map[string]diff{"72558007": {ID: "72558007"}}, + }, + { + name: "empty array result is an empty map", + status: http.StatusOK, + body: `{"result": [], "error_code": null, "error_info": null}`, + }, + { + name: "null result", + status: http.StatusOK, + body: `{"result": null, "error_code": null, "error_info": null}`, + }, + { + name: "invalid token is a conduit error, not a status error", + status: http.StatusOK, + body: `{"result": null, "error_code": "ERR-INVALID-SESSION", "error_info": "Session key is not present."}`, + wantErr: true, + wantErrorCode: "ERR-INVALID-SESSION", + }, + { + name: "conduit error without info", + status: http.StatusOK, + body: `{"result": null, "error_code": "ERR-CONDUIT-CORE"}`, + wantErr: true, + wantErrorCode: "ERR-CONDUIT-CORE", + }, + { + name: "non-200 reply is a status error", + status: http.StatusBadGateway, + body: `upstream unavailable`, + wantErr: true, + wantStatus: http.StatusBadGateway, + }, + { + name: "non-JSON body", + status: http.StatusOK, + body: `maintenance`, + wantErr: true, + }, + { + name: "result of the wrong shape", + status: http.StatusOK, + body: `{"result": "72558007", "error_code": null}`, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client, _ := newTestClient(t, tt.status, tt.body) + + var got map[string]diff + err := client.Call(context.Background(), "differential.querydiffs", map[string]any{"ids": []int{72558007}}, &got) + + if !tt.wantErr { + require.NoError(t, err) + assert.Equal(t, tt.want, got) + return + } + require.Error(t, err) + + var statusErr *phttp.StatusError + if tt.wantStatus != 0 { + require.True(t, errors.As(err, &statusErr)) + assert.Equal(t, tt.wantStatus, statusErr.StatusCode) + } else { + assert.False(t, errors.As(err, &statusErr)) + } + + var conduitErr *Error + if tt.wantErrorCode != "" { + require.True(t, errors.As(err, &conduitErr)) + assert.Equal(t, tt.wantErrorCode, conduitErr.Code) + } else { + assert.False(t, errors.As(err, &conduitErr)) + } + }) + } +} + +func TestClient_Call_HonorsContext(t *testing.T) { + client, fake := newTestClient(t, http.StatusOK, `{"result": {}}`) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := client.Call(ctx, "conduit.ping", nil, nil) + + require.ErrorIs(t, err, context.Canceled) + assert.Empty(t, fake.received()) +}