Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions platform/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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/`.
Expand Down
20 changes: 20 additions & 0 deletions platform/conduit/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -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",
],
)
47 changes: 47 additions & 0 deletions platform/conduit/README.md
Original file line number Diff line number Diff line change
@@ -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/<method>` 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.
145 changes: 145 additions & 0 deletions platform/conduit/conduit.go
Original file line number Diff line number Diff line change
@@ -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")
}
Loading