client.go

// Legacy Materials in this file remain under their original licenses.
// See LEGACY_NOTICES.md.

// Modifications Copyright (c) 2026 Super Durable, Inc.
//
// Modifications after the Legacy Cutoff are licensed under the
// Sustainable Use License 1.0.
// Legacy Materials remain under their original licenses.
// See LICENSE and LEGACY_NOTICES.md.

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/superdurable/dex/sdk-go/dex"
	"github.com/superdurable/dex/sdk-go/dex/ptr"
)

type UpdateOrderInput struct {
	Status string
}

type UpdateOrderOutput struct {
	Status string
}

func (OrderFlow) UpdateOrder(
	ctx dex.Context,
	input UpdateOrderInput,
) (*dex.RPCResult[UpdateOrderOutput], error) {
	if err := OrderStatus.Set(ctx, input.Status); err != nil {
		return nil, err
	}
	return &dex.RPCResult[UpdateOrderOutput]{
		Output: UpdateOrderOutput{Status: input.Status},
	}, nil
}

var _ dex.RPC[UpdateOrderInput, UpdateOrderOutput] = Orders.UpdateOrder

// startOrder retries one start with the same startAttempt; a rerun after a failed run uses the next one.
func startOrder(
	ctx context.Context,
	client *dex.Client,
	flowID string,
	startAttempt int,
) (string, error) {
	initialStatus, err := dex.InitialAttribute(OrderStatus, "created")
	if err != nil {
		return "", err
	}
	initialQuantity, err := dex.InitialAttributeMapValue(
		ItemQuantities,
		"sku-1",
		2,
	)
	if err != nil {
		return "", err
	}

	return client.StartFlow(
		ctx,
		Orders,
		flowID,
		OrderInput{OrderID: flowID},
		dex.StartFlowOptions{
			Timeout:       ptr.Any(24 * time.Hour),
			IDReusePolicy: dex.IDReuseAllowIfPreviousFailed,
			StartDelay:    ptr.Any(5 * time.Second),
			RetryPolicy: &dex.FlowRetryPolicy{
				InitialInterval:    time.Second,
				BackoffCoefficient: 2,
				MaximumInterval:    time.Minute,
				MaximumAttempts:    3,
			},
			Attributes: []dex.InitialAttributeDef{
				initialStatus,
				initialQuantity,
			},
			ConfigOverride: &dex.FlowConfig{
				ActiveStepSearchMode: ptr.Any(dex.SearchAllActiveSteps),
				StepDurability:       ptr.Any(dex.StepDurabilitySync),
			},
			AlreadyStarted: &dex.AlreadyStartedOptions{
				IgnoreError: true,
			},
			RequestID: ptr.Any(fmt.Sprintf("start-order/%s/%d", flowID, startAttempt)),
		},
	)
}

func writeOrderProgress(
	ctx context.Context,
	client *dex.Client,
	flowID string,
) error {
	return client.WriteStream(
		ctx,
		flowID,
		OrderProgress,
		"frontend/started",
		"waiting for inventory",
	)
}

func readOrderProgress(
	ctx context.Context,
	client *dex.Client,
	flowID string,
	resumeToken string,
) (string, dex.StreamMessage, error) {
	var progress string
	message, err := client.ReadStream(
		ctx,
		flowID,
		OrderProgress,
		resumeToken,
		&progress,
	)
	return progress, message, err
}

func invokeUpdateOrder(
	ctx context.Context,
	client *dex.Client,
	flowID string,
) (UpdateOrderOutput, error) {
	var output UpdateOrderOutput
	err := client.InvokeRPC(
		ctx,
		flowID,
		Orders.UpdateOrder,
		UpdateOrderInput{Status: "processing"},
		&output,
	)
	return output, err
}

func waitForOrderStatus(
	ctx context.Context,
	client *dex.Client,
	flowID string,
) error {
	var matched string
	return client.WaitForAttributeMatch(
		ctx,
		flowID,
		OrderStatus,
		dex.AttributeMatchEqual("shipped"),
		&matched,
		dex.WaitForAttributeOptions{RequestID: uuid.NewString(), RequestTimeout: 10 * time.Second},
	)
}

func waitForItemQuantity(
	ctx context.Context,
	client *dex.Client,
	flowID string,
) error {
	var matched int
	return client.WaitForAttributeMapInstanceMatch(
		ctx,
		flowID,
		ItemQuantities,
		"sku-1",
		dex.AttributeMatchEqual(3),
		&matched,
		dex.WaitForAttributeOptions{RequestID: uuid.NewString(), RequestTimeout: 10 * time.Second},
	)
}

func stopOrder(
	ctx context.Context,
	client *dex.Client,
	flowID string,
) error {
	return client.StopFlow(
		ctx,
		flowID,
		dex.StopOptions{
			Type:   dex.CancelFlow,
			Reason: "customer requested cancellation",
		},
	)
}

func waitForOrder(
	ctx context.Context,
	client *dex.Client,
	flowID string,
) (dex.FlowResult, error) {
	return client.WaitForFlow(
		ctx,
		flowID,
		dex.WaitForFlowOptions{
			NeedsResults: true,
		},
	)
}

func searchOrders(
	ctx context.Context,
	client *dex.Client,
) (dex.SearchFlowsPage, error) {
	return client.SearchFlows(
		ctx,
		"order-status = 'shipped'",
		100,
		"",
	)
}

func timeTravelOrder(
	ctx context.Context,
	client *dex.Client,
	flowID string,
) (string, error) {
	return client.TimeTravel(
		ctx,
		flowID,
		dex.TimeTravelOptions{
			Type:   dex.TimeTravelToBeginning,
			Reason: "reprocess order",
		},
	)
}

func skipOrderTimer(
	ctx context.Context,
	client *dex.Client,
	flowID string,
) error {
	return client.SkipTimer(
		ctx,
		flowID,
		dex.StepExecutionID{
			StepType: dex.GetFinalStepType(WaitForCommand),
		},
		dex.TimerID{ConditionID: "timeout"},
	)
}

func updateOrderConfig(
	ctx context.Context,
	client *dex.Client,
	flowID string,
) error {
	return client.UpdateFlowConfig(
		ctx,
		flowID,
		dex.FlowConfig{
			StepDurability: ptr.Any(dex.StepDurabilityAsync),
		},
	)
}

func waitForOrderStep(
	ctx context.Context,
	client *dex.Client,
	flowID string,
) error {
	return client.WaitForStepCompletion(
		ctx,
		flowID,
		dex.StepExecutionID{
			StepType: dex.GetFinalStepType(WaitForCommand),
		},
		dex.WaitForStepCompletionOptions{RequestID: uuid.NewString(), RequestTimeout: 10 * time.Second},
	)
}

func checkHealth(
	ctx context.Context,
	client *dex.Client,
) (dex.HealthInfo, error) {
	return client.HealthCheck(ctx)
}

func decodeOrderStatus(values map[string]dex.Value) (string, error) {
	value, found := values[OrderStatus.AttributeName()]
	if !found {
		return "", fmt.Errorf("order status is missing")
	}
	var status string
	if err := value.Decode(&status); err != nil {
		return "", err
	}
	return status, nil
}