stephttp

README

Inngest API Functions

This package enables using Inngest's step tooling within synchronous API handlers, providing full observability and tracing for your HTTP endpoints.

Features

• Full step observability: Every step.Run() call is automatically traced and logged
• Background checkpointing: Step data is sent to Inngest in the background without blocking your API
• Seamless integration: Use existing step.Run() functions in API handlers
• APM out of the box: Get metrics, traces, and monitoring for every API endpoint
• Event triggering: Send events from API handlers that trigger async workflows

Documentation

Index

• func Configure(ctx context.Context, opts FnOpts)
• func Setup(opts SetupOpts) *provider
• func UpdateOmitResponseBody(ctx context.Context, to bool)
• type APIClient
  • func NewAPIClient(baseURL, primaryKey, fallbackKey string) *APIClient
  • func (c *APIClient) CheckpointNewRun(ctx context.Context, runID ulid.ULID, input NewAPIRunData, ...) (*CheckpointRun, error)
  • func (c *APIClient) CheckpointResponse(ctx context.Context, run CheckpointRun, result APIResult) error
  • func (c *APIClient) CheckpointSteps(ctx context.Context, run CheckpointRun, steps []sdkrequest.GeneratorOpcode) error
  • func (c *APIClient) GetSteps(ctx context.Context, runID ulid.ULID) (map[string]json.RawMessage, error)
• type APIResult
• type AsyncResponse
• type AsyncResponseCustom
• type AsyncResponseRedirect
• type AsyncResponseToken
• type CheckpointNewRunRequest
• type CheckpointRun
• type FnOpts
• type NewAPIRunData
• type OptionalSetupOpts
• type Provider
• type SetupOpts

Functions

func Configure

func Configure(ctx context.Context, opts FnOpts)

func Setup

func Setup(opts SetupOpts) *provider

Setup creates a new API provider instance

func UpdateOmitResponseBody

func UpdateOmitResponseBody(ctx context.Context, to bool)

UpdateOmitResponseBody sets whether the response body will be tracked in logs and traces. You can call this at any time before sending the API response and this will be respected.

Types

type APIClient

type APIClient struct {
	// contains filtered or unexported fields
}

APIClient handles HTTP requests to the checkpoint API

func NewAPIClient

func NewAPIClient(baseURL, primaryKey, fallbackKey string) *APIClient

NewAPIClient creates a new API client with the given domain and signing keys

func (*APIClient) CheckpointNewRun

func (c *APIClient) CheckpointNewRun(ctx context.Context, runID ulid.ULID, input NewAPIRunData, steps ...sdkrequest.GeneratorOpcode) (*CheckpointRun, error)

CheckpointNewRun creates a new API run checkpoint

func (*APIClient) CheckpointResponse

func (c *APIClient) CheckpointResponse(ctx context.Context, run CheckpointRun, result APIResult) error

CheckpointResponse saves the final API response

func (*APIClient) CheckpointSteps

func (c *APIClient) CheckpointSteps(ctx context.Context, run CheckpointRun, steps []sdkrequest.GeneratorOpcode) error

CheckpointSteps saves step execution state

func (*APIClient) GetSteps

func (c *APIClient) GetSteps(ctx context.Context, runID ulid.ULID) (map[string]json.RawMessage, error)

type APIResult

type APIResult struct {
	// StatusCode represents the status code for the API result
	StatusCode int `json:"status_code"`
	// Headers represents any response headers sent in the server response
	Headers map[string]string `json:"headers"`
	// Body represents the API response.  This may be nil by default.  It is only
	// captured when you manually specify that you want to track the result.
	Body []byte `json:"body,omitempty"`
	// Duration represents the duration
	Duration time.Duration `json:"duration"`
	// Error represents any error from the API.  This is only for internal errors,
	// eg. when a step permanently fails
	Error string `json:"error,omitempty"`
}

APIResult represents the final result of an API function call

type AsyncResponse

type AsyncResponse interface {
	// contains filtered or unexported methods
}

type AsyncResponseCustom

type AsyncResponseCustom http.HandlerFunc

AsyncResponseCustom allows you to configure custom HTTP responses when a sync function hits an async step.

When you use eg. `step.sleep`, the function pauses and you must return a response to the user.

type AsyncResponseRedirect

type AsyncResponseRedirect struct {
	// URL is the optional URL to redirect to.  If empty, we will redirect to
	// the Inngest API which will automatically block until the function completes.
	//
	// Note that this accepts a token which can be used to hit the Inngest API to
	// block for the API result.
	URL func(runID ulid.ULID, token string) string
}

AsyncResponseRedirect redirects the user to a URL which will block until the async function completes, then return the output to the user.

This is seamless, and as long as the function executes within a specific period the user will not know that the API finished asynchronously.

type AsyncResponseToken

type AsyncResponseToken struct{}

AsyncResponseToken responds with the Token format, allowing clients to wait for the function to finish by hitting our V2 API with the given token.

type CheckpointNewRunRequest

type CheckpointNewRunRequest struct {
	// RunID represents the run ID for this request.  This is generated on the
	// client, and embeds the timestamp that the request started
	RunID ulid.ULID `json:"run_id"`

	// Idempotency allows the customization of an idempotency key, allowing us to
	// handle API idempotency using Inngest.
	Idempotency string `json:"idempotency"`

	// Event embeds the key request information which is used as the triggering
	// event for API-based runs.
	Event inngestgo.GenericEvent[NewAPIRunData] `json:"event"`

	// Steps are optional steps to send when creating the request.
	Steps []sdkrequest.GeneratorOpcode `json:"steps,omitempty,omitzero"`
}

CheckpointNewRunRequest represents the entire request payload used to create new API-based runs.

type CheckpointRun

type CheckpointRun struct {
	// FnID represents the ID of the function that the checkpoint run relates to.
	// This is required to be passed back in future step and response checkpoint calls
	// for proper tracking.
	FnID uuid.UUID `json:"fn_id"`
	// AppID represents the ID of the app that the checkpoint run relates to.
	// This is required to be passed back in future step and response checkpoint calls
	// for proper tracking.
	AppID uuid.UUID `json:"app_id"`
	// RunID is the function run ID created for this execution.
	RunID ulid.ULID `json:"run_id"`
	// Token is the token to use when redirecting if this is an async checkpoint.
	Token string `json:"token,omitempty"`

	// Stack is the current stack, used when resuming requests.
	Stack []string
	// Signature is the signature used when resuming requests.
	Signature string
	// Attempt is the attempt number
	Attempt int
}

CheckpointRun represents the response payload for a successful run creation, and is used to resume any checkpointed run (async or sync).

type FnOpts

type FnOpts struct {
	// ID represents the function ID.  You should always set this, and must
	// set this when the URL contains values or identifiers, eg /users/123.
	ID string

	// OmitRequestBody prevents the incoming request from being stored every time.
	OmitRequestBody bool

	// OmitResponseBody prevents the API response from being stored every time.
	//
	// Note that you can override this by calling `stephttp.UpdateOmitResponseBody(ctx, bool)`
	// before the API responds to dynamically adjust whether the response is stored. This
	// allows you to properly store eg. errors for debugging, then only omit successful
	// responses at the end of your function.
	OmitResponseBody bool

	// AsyncResponse determines how we respond to a user when an API hits an
	// async step (eg. step.sleep, step.waitForEvent) or if a step errors.
	//
	// Mot of the time, AsyncResponseRedirect allows for seamless handling of
	// step errors and sleeping steps.
	//
	// For more information, see the Inngest docs.
	//
	// By default, this uses the AsyncResponseRedirect response type, redirecting
	// the user to a URL that will block and return the API result automatically
	// once the API finishes.
	AsyncResponse AsyncResponse
}

FnOpts allows you to define function configuration options for your API-based Inngest function.

type NewAPIRunData

type NewAPIRunData struct {
	// Domain is the domain that served the incoming request.
	Domain string `json:"domain"`
	// Method is the incoming request method.  This is used for RESTful
	// API endpoints.
	Method string `json:"method"`
	// Path is the path for the incoming request.
	Path string `json:"path"` // request path
	// Fn is the optional function slug.  If not present, this is created
	// using a combination of the method and the path: "POST /v1/runs"
	Fn string `json:"fn"`

	// IP is the IP that created the request.
	IP string `json:"ip"` // incoming IP
	// ContentType is the content type for the request.
	ContentType string `json:"content_type"`
	// QueryParams are the query parameters for the request, as a single string
	// without the leading "?".
	//
	// NOTE: This is optional;  we do not require that users store the query params
	// for every request, as this may contain data that users choose not to log.
	QueryParams string `json:"query_params"`
	// Body is the incoming request body.
	//
	// NOTE: This is optional;  we do not require that users store the body for
	// every request, as this may contain data that users choose not to log.
	Body []byte `json:"body"`
}

NewAPIRunData represents event data stored and used to create new API-based runs.

type OptionalSetupOpts

type OptionalSetupOpts struct {
	// TrackAllEndpoints, if set to true, will track all requests to all API endpoints,
	// even if they don't use steps.
	//
	// By default, only API endpoints that use steps will be tracked.
	TrackAllEndpoints bool

	// DefaultAsyncResponse defines the default async response type.  Each function
	// can override the async repsonse type using function configuration.
	DefaultAsyncResponse AsyncResponse

	// SigningKey is the Inngest signing key for authentication.  If empty, this defaults
	// to os.Getenv("INNGEST_SIGNING_KEY").
	SigningKey string
	// SigningKeyFallback is the optional signing key fallback. If empty, this defaults
	// to os.Getenv("INNGEST_SIGNING_KEY_FALLBACK").
	SigningKeyFallback string
	// BaseURL is the URL of the Inngest API.  If empty, this:
	//
	//   1. Checks to see if INNGEST_DEV is set, indicating dev mode.  If set, we
	//      attempt to use the INNGEST_DEV env var as the base URL if set to a URL,
	//      or default to "http://127.0.0.1:8288" for dev mode.
	//   2. If INNGEST_DEV is not set, we default to the production URL:
	//      "https://api.inngest.com".
	BaseURL string
	// Middleware represents optional middleware to run before and after processing.
	Middleware []func() middleware.Middleware
}

type Provider

type Provider interface {
	// ServeHTTP is the middleware that allows the Inngest handler to work.
	ServeHTTP(next http.HandlerFunc) http.HandlerFunc

	// Middleware returns stdlib-style middleware to wrap other HTTP handlers
	Middleware(next http.Handler) http.Handler

	// Wait provides a mechanism to wait for all cehckpoints to finish before shutting down.
	// Cancel the incoming context to quit polling for checkpoint progres.
	Wait(ctx context.Context) chan bool
}

type SetupOpts

type SetupOpts struct {
	// Domain is the domain for this API (e.g., "api.mycompany.com")
	Domain string
	// Optional represents optional setup options that you can confgure.
	Optional OptionalSetupOpts
}

SetupOpts contains configuration for the API middleware. Optional configuration is supplied via SetupOpt adapters.