Write a FlowDSL Node in Go

package main

import (
    "context"
    "fmt"

    flowdsl "github.com/flowdsl/flowdsl-go"
)

// FilterNode implements the filter_by_priority operation.
// It reads the "priority" field from the input packet and routes
// P0/P1 events to "urgent_out", P2+ to "normal_out".
type FilterNode struct {
    urgentPriorities map[string]bool
}

// OperationID returns the snake_case identifier that matches the
// operationId in the FlowDSL document.
func (n *FilterNode) OperationID() string {
    return "filter_by_priority"
}

// Init is called once at startup with the node's static settings.
func (n *FilterNode) Init(settings flowdsl.Settings) error {
    urgentList, _ := settings.GetStringSlice("urgentPriorities")
    if len(urgentList) == 0 {
        urgentList = []string{"P0", "P1"}
    }
    n.urgentPriorities = make(map[string]bool, len(urgentList))
    for _, p := range urgentList {
        n.urgentPriorities[p] = true
    }
    return nil
}

// Handle is called once per incoming packet.
func (n *FilterNode) Handle(ctx context.Context, input flowdsl.NodeInput) (flowdsl.NodeOutput, error) {
    // Read the input packet from the "in" port
    payload, err := input.Packet("in")
    if err != nil {
        return flowdsl.NodeOutput{}, fmt.Errorf("filter_by_priority: reading input: %w", err)
    }

    // Extract the priority field
    priority, ok := payload.GetString("priority")
    if !ok {
        // Missing priority — treat as normal
        priority = "P2"
    }

    // Route based on priority
    if n.urgentPriorities[priority] {
        return flowdsl.NodeOutput{}.Send("urgent_out", payload), nil
    }
    return flowdsl.NodeOutput{}.Send("normal_out", payload), nil
}

package main

import (
    "log/slog"
    "os"

    flowdsl "github.com/flowdsl/flowdsl-go"
)

func main() {
    logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
        Level: slog.LevelInfo,
    }))

    server := flowdsl.NewNodeServer(
        flowdsl.WithLogger(logger),
        flowdsl.WithGRPCPort(50051),
        flowdsl.WithManifestFile("flowdsl-node.json"),
    )

    server.Register(&FilterNode{})

    logger.Info("starting filter-node gRPC server", "port", 50051)
    if err := server.ServeGRPC(); err != nil {
        logger.Error("server failed", "error", err)
        os.Exit(1)
    }
}

func (n *FilterNode) Handle(ctx context.Context, input flowdsl.NodeInput) (flowdsl.NodeOutput, error) {
    payload, err := input.Packet("in")
    if err != nil {
        // Return a typed FlowDSL error for proper dead letter categorization
        return flowdsl.NodeOutput{}, flowdsl.NewNodeError(
            flowdsl.ErrCodeInputMissing,
            "missing input packet on port 'in'",
            err,
        )
    }

    priority, ok := payload.GetString("priority")
    if !ok {
        // Return a validation error — this will NOT be retried (it's a data problem)
        return flowdsl.NodeOutput{}, flowdsl.NewNodeError(
            flowdsl.ErrCodeValidation,
            "priority field missing from payload",
            nil,
        )
    }

    // ... routing logic
}

func (n *FilterNode) Handle(ctx context.Context, input flowdsl.NodeInput) (flowdsl.NodeOutput, error) {
    // Extract FlowDSL trace context for correlated logging
    traceCtx := flowdsl.TraceFromContext(ctx)

    logger := slog.With(
        "flowId", traceCtx.FlowID,
        "executionId", traceCtx.ExecutionID,
        "nodeId", "FilterByPriority",
    )

    payload, err := input.Packet("in")
    if err != nil {
        logger.Error("failed to read input packet", "error", err)
        return flowdsl.NodeOutput{}, err
    }

    priority, _ := payload.GetString("priority")
    logger.Info("routing packet", "priority", priority)

    // ...
}