Handle sync redirecting (#180)

This commit introduces full redirect support for sync -> async
conversion in API-based runs

Author: tonyhb

stephttp/api_client.go

@@ -29,6 +29,11 @@ type CheckpointRun struct {
 	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"`
+
+	// NOTE: The below are not included in the API response when checkpointing
+	// new runs.
 
 	// Stack is the current stack, used when resuming requests.
 	Stack []string

stephttp/configure.go

@@ -88,7 +88,7 @@ type AsyncResponseRedirect struct {
 	//
 	// Note that this accepts a token which can be used to hit the Inngest API to
 	// block for the API result.
-	URL func(token string) string
+	URL func(runID ulid.ULID, token string) string
 }
 
 func (a AsyncResponseRedirect) isAsyncResponse() {}

stephttp/request_owner.go

@@ -131,8 +131,8 @@ func (o *requestOwner) handle(ctx context.Context) error {
 
 	if sdkrequest.HasAsyncOps(o.mgr.Ops(), o.run.Attempt, 0) {
 		// Always checkpoint first, then handle the async conversion.
-		o.handleFirstCheckpoint(ctx)
-		return o.handleAsyncConversion(ctx)
+		token := o.handleFirstCheckpoint(ctx)
+		return o.handleAsyncConversion(ctx, token)
 	}
 
 	// Attempt to flush the response directly to the client immediately, reducing TTFB
@@ -169,7 +169,7 @@ func (o *requestOwner) handle(ctx context.Context) error {
 //
 // We also need to handle the API response to our user, which is either a token,
 // a redirect, or a custom response.
-func (o *requestOwner) handleAsyncConversion(ctx context.Context) error {
+func (o *requestOwner) handleAsyncConversion(ctx context.Context, token string) error {
 	if !sdkrequest.HasAsyncOps(o.mgr.Ops(), o.run.Attempt, 0) {
 		return nil
 	}
@@ -193,19 +193,19 @@ func (o *requestOwner) handleAsyncConversion(ctx context.Context) error {
 	case AsyncResponseToken:
 		return json.NewEncoder(o.w).Encode(asyncResponseToken{
 			RunID: o.run.RunID,
-			Token: redirectToken(o.run.RunID),
+			Token: token,
 		})
 	case AsyncResponseCustom:
 		v(o.w, o.r)
 		return nil
 	case AsyncResponseRedirect:
 		if v.URL != nil {
-			url = v.URL(redirectToken(o.run.RunID))
+			url = v.URL(o.run.RunID, token)
 		}
 	}
 
 	if url == "" {
-		url = defaultRedirectURL(o.provider.opts, o.run.RunID)
+		url = defaultRedirectURL(o.provider.opts, o.run.RunID, token)
 	}
 
 	http.Redirect(o.w, o.r, url, http.StatusSeeOther)
@@ -312,7 +312,9 @@ func (o *requestOwner) call(ctx context.Context) APIResult {
 // It also checkpoints the first N steps (potentially including the entire function).
 //
 // This is a blocking operation;  to run this in the background use a goroutine.
-func (o *requestOwner) handleFirstCheckpoint(ctx context.Context) {
+//
+// This returns an optional token used when redirecting to async outputs.
+func (o *requestOwner) handleFirstCheckpoint(ctx context.Context) string {
 	var (
 		requestBody []byte
 		err         error
@@ -357,10 +359,11 @@ func (o *requestOwner) handleFirstCheckpoint(ctx context.Context) {
 	}, o.mgr.Ops()...)
 	if err != nil {
 		o.provider.logger.Error("error creating new api-based inngest run", "error", err, "run_id", o.run.RunID)
-		return
+		return ""
 	}
 
 	o.run = *resp
+	return resp.Token
 }
 
 // validateResumeRequestSignature validates the signature for resume requests.

stephttp/util.go

@@ -3,6 +3,7 @@ package stephttp
 import (
 	"bufio"
 	"bytes"
+	"fmt"
 	"io"
 	"net"
 	"net/http"
@@ -11,13 +12,13 @@ import (
 	"github.com/oklog/ulid/v2"
 )
 
-func redirectToken(runID ulid.ULID) string {
-	// TODO: SIGN WITH A KEY OR RANDOM TOKEN
-	return runID.String()
-}
-
-func defaultRedirectURL(o SetupOpts, runID ulid.ULID) string {
-	return o.baseURL() + "/v2/public/runs/" + redirectToken(runID)
+func defaultRedirectURL(o SetupOpts, runID ulid.ULID, token string) string {
+	return fmt.Sprintf(
+		"%s/v1/http/runs/%s/output?token=%s",
+		o.baseURL(),
+		runID,
+		token,
+	)
 }
 
 // responseWriter captures the response for storing as the API result
@@ -56,10 +57,10 @@ func (rw *responseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
 	if !ok {
 		return nil, nil, http.ErrNotSupported
 	}
-	
+
 	// Mark as hijacked so we stop capturing response data
 	rw.hijacked = true
-	
+
 	return hijacker.Hijack()
 }