Durable Workflows in Go

import (
	"context"
	"errors"
	"time"

	"github.com/inngest/inngestgo"
	"github.com/inngest/inngestgo/step"
)

type OrderPlaced struct {
	OrderID string `json:"orderId"`
}

type OrderCancelled struct {
	OrderID string `json:"orderId"`
}

func ProcessOrder(client inngestgo.Client) (inngestgo.ServableFunction, error) {
	return inngestgo.CreateFunction(
		client,
		inngestgo.FunctionOpts{ID: "process-order"},
		// any time the `shop/order.placed` event is received, this will run.
		inngestgo.EventTrigger("shop/order.placed", nil),
		func(ctx context.Context, input inngestgo.Input[OrderPlaced]) (any, error) {
			order, err := step.Run(ctx, "load-order", func(ctx context.Context) (Order, error) {
				return loadOrder(ctx, input.Event.Data.OrderID)
			})
			if err != nil {
				return nil, err
			}
			if _, err := step.Run(ctx, "charge-order", func(ctx context.Context) (any, error) {
				return nil, chargeOrder(ctx, order)
			}); err != nil {
				return nil, err
			}
			if _, err := step.Run(ctx, "send-receipt", func(ctx context.Context) (any, error) {
				return nil, sendReceipt(ctx, order)
			}); err != nil {
				return nil, err
			}

			// this will wait for the `shop/order.cancelled` event for up to 6 hours,
			// and resume immediately when a matching event is received.  If an event
			// isn't received within 6 hours, the function resumes and WaitForEvent
			// returns step.ErrEventNotReceived.
			//
			// your compute is *not running* during this wait, and you do not need to handle
			// any matching.
			_, err = step.WaitForEvent[inngestgo.GenericEvent[OrderCancelled]](
				ctx,
				"wait-for-cancellation",
				step.WaitForEventOpts{
					Event:   "shop/order.cancelled",
					If:      inngestgo.StrPtr("event.data.orderId == async.data.orderId"),
					Timeout: 6 * time.Hour,
				},
			)
			if errors.Is(err, step.ErrEventNotReceived) {
				return map[string]any{"orderId": order.ID}, nil
			}
			if err != nil {
				return nil, err
			}

			// the order was cancelled, as we received a cancel event
			if _, err := step.Run(ctx, "refund-order", func(ctx context.Context) (any, error) {
				return nil, refundOrder(ctx, order)
			}); err != nil {
				return nil, err
			}
			return map[string]any{"orderId": order.ID, "status": "cancelled"}, nil
		},
	)
}

import (
	"log"
	"net/http"

	"github.com/inngest/inngestgo"
)

func Serve() {
	client, err := inngestgo.NewClient(inngestgo.ClientOpts{AppID: "billing"})
	if err != nil {
		log.Fatal(err)
	}

	// Create your functions with inngestgo.CreateFunction(client, ...) before
	// serving.  Each one registers itself with the client.

	// Inngest calls this endpoint to run your functions.
	http.Handle("/api/inngest", client.Serve())
	log.Fatal(http.ListenAndServe(":8080", nil))
}

import (
	"context"
	"log"
	"os/signal"
	"syscall"

	"github.com/inngest/inngestgo"
)

func Worker() {
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	client, err := inngestgo.NewClient(inngestgo.ClientOpts{AppID: "billing"})
	if err != nil {
		log.Fatal(err)
	}

	// Create your functions with inngestgo.CreateFunction(client, ...) before
	// connecting.  Each one registers itself with the client.

	// Connect opens a WebSocket to Inngest and returns once it's active.
	conn, err := inngestgo.Connect(ctx, inngestgo.ConnectOpts{
		Apps: []inngestgo.Client{client},
	})
	if err != nil {
		log.Fatal(err)
	}

	// Close finishes in-flight steps before it returns.
	<-ctx.Done()
	if err := conn.Close(); err != nil {
		log.Fatal(err)
	}
}

import (
	"context"
	"time"

	"github.com/inngest/inngestgo"
	"github.com/inngest/inngestgo/step"
)

type AccountUpdated struct {
	AccountID string `json:"accountId"`
	Plan      string `json:"plan"`
}

func SyncAccount(client inngestgo.Client) (inngestgo.ServableFunction, error) {
	return inngestgo.CreateFunction(
		client,
		inngestgo.FunctionOpts{ID: "sync-account"},
		inngestgo.MultipleTriggers{
			// Run for each account update on a paid plan.
			inngestgo.EventTrigger("account/updated", inngestgo.StrPtr("event.data.plan != 'free'")),
			// Also run every night, spread across up to 10 minutes.
			inngestgo.CronTriggerWithJitter("TZ=America/New_York 0 2 * * *", 10*time.Minute),
		},
		func(ctx context.Context, input inngestgo.Input[AccountUpdated]) (any, error) {
			// A cron run has no event data, so AccountID is empty.
			_, err := step.Run(ctx, "sync-account", func(ctx context.Context) (any, error) {
				return nil, syncAccount(ctx, input.Event.Data.AccountID)
			})
			return nil, err
		},
	)
}

import (
	"context"
	"fmt"
	"time"

	"github.com/inngest/inngestgo"
	inngesterrors "github.com/inngest/inngestgo/errors"
	"github.com/inngest/inngestgo/step"
)

type ReportRequested struct {
	ReportID string `json:"reportId"`
}

var errRateLimited = fmt.Errorf("rate limited")

func BuildReport(client inngestgo.Client) (inngestgo.ServableFunction, error) {
	return inngestgo.CreateFunction(
		client,
		inngestgo.FunctionOpts{ID: "build-report"},
		inngestgo.EventTrigger("report/requested", nil),
		func(ctx context.Context, input inngestgo.Input[ReportRequested]) (any, error) {
			if input.Event.Data.ReportID == "" {
				// The event can never succeed, so don't retry.
				return nil, inngesterrors.NoRetryError(fmt.Errorf("reportId is required"))
			}

			return step.Run(ctx, "build-report", func(ctx context.Context) (Report, error) {
				report, err := buildReport(ctx)
				if err == errRateLimited {
					// Retry this step in 5 minutes instead of on the default backoff.
					return Report{}, inngesterrors.RetryAtError(err, time.Now().Add(5*time.Minute))
				}
				// Any other error retries the step with backoff.
				return report, err
			})
		},
	)
}

import (
	"context"
	"time"

	"github.com/inngest/inngestgo"
	"github.com/inngest/inngestgo/step"
)

type InvoiceCreated struct {
	InvoiceID string `json:"invoiceId"`
	AccountID string `json:"accountId"`
}

func ChargeInvoice(client inngestgo.Client) (inngestgo.ServableFunction, error) {
	return inngestgo.CreateFunction(
		client,
		inngestgo.FunctionOpts{
			ID:   "charge-invoice",
			Name: "Charge invoice",
			// Retry each failed step up to 10 times.
			Retries: inngestgo.IntPtr(10),
			// Run one invoice at a time for each account.
			Concurrency: &inngestgo.ConfigConcurrency{
				Fn: []inngestgo.ConfigFnConcurrency{
					{Limit: 1, Key: inngestgo.StrPtr("event.data.accountId")},
				},
			},
			// Start at most 100 runs a minute.
			Throttle: &inngestgo.ConfigThrottle{Limit: 100, Period: time.Minute},
			// Cancel the run when the invoice is voided.
			Cancel: []inngestgo.ConfigCancel{
				{Event: "invoice/voided", If: inngestgo.StrPtr("event.data.invoiceId == async.data.invoiceId")},
			},
			// Cancel the run if it doesn't finish within an hour.
			Timeouts: &inngestgo.ConfigTimeouts{Finish: inngestgo.Ptr(time.Hour)},
		},
		inngestgo.EventTrigger("invoice/created", nil),
		func(ctx context.Context, input inngestgo.Input[InvoiceCreated]) (any, error) {
			invoice, err := step.Run(ctx, "load-invoice", func(ctx context.Context) (Invoice, error) {
				return loadInvoice(ctx, input.Event.Data.InvoiceID)
			})
			if err != nil {
				return nil, err
			}
			_, err = step.Run(ctx, "charge", func(ctx context.Context) (any, error) {
				return nil, chargeInvoice(ctx, invoice)
			})
			return nil, err
		},
	)
}