refactor: replace Handler with Proc/Func pattern

Upgrade API to support both fire-and-forget (Proc) and request-response
(Func) handler patterns with explicit Replier interface for response
handling.

- Rename Handler[T] to Proc[T] with Run method
- Add Func[T, R] interface with Call method for request-response
- Rename Parsed to Message with Replier instead of Complete callback
- Add Replier interface with Reply/Fail methods
- Update registration: RegisterProc, RegisterFunc, RegisterProcFunc,
  RegisterFuncFunc
- Use sourceRef struct for more efficient adaptive ordering
- Update all documentation to reflect new API

Author: bjaus

README.md

@@ -12,8 +12,9 @@ A flexible message routing framework for event-driven Go applications.
 - **Multi-Source Routing** — Route messages from webhooks, message queues, or custom formats through a single processor
 - **Discriminator Pattern** — Cheap detection before expensive parsing for O(1) hot-path matching
 - **Typed Handlers** — Automatic JSON unmarshaling and validation with generics
+- **Proc/Func Pattern** — Fire-and-forget procedures or request-response functions
+- **Replier Interface** — Built-in support for request-response transports (Step Functions, etc.)
 - **Pluggable Hooks** — Observability without coupling to specific logging or metrics systems
-- **Completion Callbacks** — Built-in support for async acknowledgment patterns
 - **Format Agnostic** — Inspector/View abstraction supports JSON, protobuf, or custom formats
 - **Zero Allocation Matching** — Uses gjson for efficient JSON field lookups
 
@@ -44,11 +45,11 @@ type UserCreatedPayload struct {
     Email  string `json:"email"`
 }
 
-// Define your handler
-type UserCreatedHandler struct{}
+// Define a procedure (fire-and-forget)
+type UserCreatedProc struct{}
 
-func (h *UserCreatedHandler) Handle(ctx context.Context, p UserCreatedPayload) error {
-    log.Printf("User created: %s (%s)", p.UserID, p.Email)
+func (p *UserCreatedProc) Run(ctx context.Context, payload UserCreatedPayload) error {
+    log.Printf("User created: %s (%s)", payload.UserID, payload.Email)
     return nil
 }
 
@@ -61,15 +62,15 @@ func (s *mySource) Discriminator() dispatch.Discriminator {
     return dispatch.HasFields("type", "payload")
 }
 
-func (s *mySource) Parse(raw []byte) (dispatch.Parsed, bool) {
+func (s *mySource) Parse(raw []byte) (dispatch.Message, error) {
     var env struct {
         Type    string          `json:"type"`
         Payload json.RawMessage `json:"payload"`
     }
-    if err := json.Unmarshal(raw, &env); err != nil || env.Type == "" {
-        return dispatch.Parsed{}, false
+    if err := json.Unmarshal(raw, &env); err != nil {
+        return dispatch.Message{}, err
     }
-    return dispatch.Parsed{Key: env.Type, Payload: env.Payload}, true
+    return dispatch.Message{Key: env.Type, Payload: env.Payload}, nil
 }
 
 func main() {
@@ -79,8 +80,8 @@ func main() {
     // Add source
     r.AddSource(&mySource{})
 
-    // Register handler
-    dispatch.Register(r, "user/created", &UserCreatedHandler{})
+    // Register procedure
+    dispatch.RegisterProc(r, "user/created", &UserCreatedProc{})
 
     // Process a message
     msg := []byte(`{"type": "user/created", "payload": {"user_id": "123", "email": "test@example.com"}}`)
@@ -98,7 +99,25 @@ The package separates concerns into three layers:
 |-------|---------------|
 | **Sources** | Parse raw bytes, extract routing key + payload |
 | **Router** | Match keys to handlers, orchestrate dispatch flow |
-| **Handlers** | Pure business logic with typed payloads |
+| **Handlers** | Pure business logic with typed payloads (Proc or Func) |
+
+### Proc vs Func
+
+The package provides two handler patterns:
+
+```go
+// Proc: Fire-and-forget (returns only error)
+type Proc[T any] interface {
+    Run(ctx context.Context, payload T) error
+}
+
+// Func: Request-response (returns result and error)
+type Func[T, R any] interface {
+    Call(ctx context.Context, payload T) (R, error)
+}
+```
+
+Use `Proc` for event handlers where you don't need to send a response. Use `Func` for request-response patterns like Step Functions tasks.
 
 ### Discriminator Pattern
 
@@ -134,6 +153,66 @@ r.AddSource(apiSource)
 r.AddGroup(protoInspector, grpcSource, kafkaSource)
 ```
 
+## Handler Registration
+
+```go
+// Register a procedure (fire-and-forget)
+dispatch.RegisterProc(r, "user/created", &UserCreatedProc{})
+
+// Register a function (request-response)
+dispatch.RegisterFunc(r, "lookup-user", &LookupUserFunc{})
+
+// Or use function adapters for simple cases
+dispatch.RegisterProcFunc(r, "ping", func(ctx context.Context, p PingPayload) error {
+    return nil
+})
+
+dispatch.RegisterFuncFunc(r, "echo", func(ctx context.Context, in Input) (*Output, error) {
+    return &Output{Value: in.Value}, nil
+})
+```
+
+## Replier Interface
+
+For transports that require sending responses back (like Step Functions), sources can provide a Replier:
+
+```go
+type Replier interface {
+    Reply(ctx context.Context, result json.RawMessage) error
+    Fail(ctx context.Context, err error) error
+}
+```
+
+Example Step Functions source:
+
+```go
+type sfnReplier struct {
+    sfn   SFNClient
+    token string
+}
+
+func (r *sfnReplier) Reply(ctx context.Context, result json.RawMessage) error {
+    return r.sfn.SendTaskSuccess(ctx, r.token, result)
+}
+
+func (r *sfnReplier) Fail(ctx context.Context, err error) error {
+    return r.sfn.SendTaskFailure(ctx, r.token, err)
+}
+
+func (s *sfnSource) Parse(raw []byte) (dispatch.Message, error) {
+    // ... parse envelope ...
+    return dispatch.Message{
+        Key:     taskType,
+        Payload: payload,
+        Replier: &sfnReplier{sfn: s.sfn, token: token},
+    }, nil
+}
+```
+
+When a Replier is present:
+- On success: router calls `Replier.Reply` with the marshaled result (or `{}` for Procs)
+- On error: router calls `Replier.Fail` with the error
+
 ## Discriminators
 
 Composable predicates for source matching:
@@ -201,33 +280,6 @@ type OnSuccessHook interface {
 }
 ```
 
-## Completion Callbacks
-
-For transports that require explicit acknowledgment after processing:
-
-```go
-func (s *taskSource) Parse(raw []byte) (dispatch.Parsed, bool) {
-    var env struct {
-        TaskID  string          `json:"task_id"`
-        Type    string          `json:"type"`
-        Payload json.RawMessage `json:"payload"`
-    }
-    if err := json.Unmarshal(raw, &env); err != nil {
-        return dispatch.Parsed{}, false
-    }
-    return dispatch.Parsed{
-        Key:     env.Type,
-        Payload: env.Payload,
-        Complete: func(ctx context.Context, err error) error {
-            if err != nil {
-                return s.taskQueue.ReportFailure(ctx, env.TaskID, err)
-            }
-            return s.taskQueue.ReportSuccess(ctx, env.TaskID)
-        },
-    }, true
-}
-```
-
 ## Validation
 
 Payloads implementing `Validate() error` are automatically validated:

dispatch.go

@@ -5,36 +5,72 @@ import (
 	"encoding/json"
 )
 
-// Handler processes a typed payload. This is the primary interface to implement.
+// Proc (procedure) processes a message without returning a result.
+// Use this for fire-and-forget patterns like event handlers.
 //
-// The type parameter T is the payload type that the handler expects. The router
-// automatically unmarshals the raw JSON payload to this type and validates it
-// if T implements validation.Validatable.
+// The type parameter T is the payload type. The router automatically
+// unmarshals JSON to T and validates it if T implements Validate() error.
 //
 // Example:
 //
-//	type UserCreatedHandler struct {
+//	type UserCreatedProc struct {
 //	    db *sql.DB
 //	}
 //
-//	func (h *UserCreatedHandler) Handle(ctx context.Context, p UserCreatedPayload) error {
-//	    _, err := h.db.ExecContext(ctx, "INSERT INTO users ...", p.UserID, p.Email)
+//	func (p *UserCreatedProc) Run(ctx context.Context, payload UserCreatedPayload) error {
+//	    _, err := p.db.ExecContext(ctx, "INSERT INTO users ...", payload.UserID)
 //	    return err
 //	}
-type Handler[T any] interface {
-	Handle(ctx context.Context, payload T) error
+type Proc[T any] interface {
+	Run(ctx context.Context, payload T) error
 }
 
-// HandlerFunc is a function adapter for Handler. Use this for simple handlers
+// ProcFunc is a function adapter for Proc. Use for simple procedures
 // that don't need a struct:
 //
-//	dispatch.RegisterFunc(r, "ping", func(ctx context.Context, p PingPayload) error {
+//	dispatch.RegisterProc(r, "user/created", func(ctx context.Context, p Payload) error {
 //	    return nil
 //	})
-type HandlerFunc[T any] func(ctx context.Context, payload T) error
+type ProcFunc[T any] func(ctx context.Context, payload T) error
 
-// Handle implements the Handler interface.
-func (f HandlerFunc[T]) Handle(ctx context.Context, payload T) error {
+// Run implements the Proc interface.
+func (f ProcFunc[T]) Run(ctx context.Context, payload T) error {
+	return f(ctx, payload)
+}
+
+// Func (function) processes a message and returns a typed result.
+// Use this for request-response patterns like Step Functions tasks.
+//
+// The type parameters are: T for input payload, R for result.
+// The router automatically unmarshals T, validates it, and marshals R.
+//
+// Example:
+//
+//	type LookupUserFunc struct {
+//	    client IdentityClient
+//	}
+//
+//	func (f *LookupUserFunc) Call(ctx context.Context, in LookupInput) (*LookupResult, error) {
+//	    user, err := f.client.GetUser(ctx, in.UserID)
+//	    if err != nil {
+//	        return nil, err
+//	    }
+//	    return &LookupResult{Email: user.Email}, nil
+//	}
+type Func[T, R any] interface {
+	Call(ctx context.Context, payload T) (R, error)
+}
+
+// FuncFunc is a function adapter for Func. Use for simple functions
+// that don't need a struct:
+//
+//	dispatch.RegisterFunc(r, "lookup-user", func(ctx context.Context, in Input) (*Result, error) {
+//	    return &Result{...}, nil
+//	})
+type FuncFunc[T, R any] func(ctx context.Context, payload T) (R, error)
+
+// Call implements the Func interface.
+func (f FuncFunc[T, R]) Call(ctx context.Context, payload T) (R, error) {
 	return f(ctx, payload)
 }
 
@@ -49,6 +85,7 @@ func (f HandlerFunc[T]) Handle(ctx context.Context, payload T) error {
 //   - SNS notifications
 //   - Step Functions task tokens
 //   - Kinesis records
+//   - SQS messages
 //   - Custom formats
 //
 // Example:
@@ -61,18 +98,15 @@ func (f HandlerFunc[T]) Handle(ctx context.Context, payload T) error {
 //	    return dispatch.HasFields("type", "payload")
 //	}
 //
-//	func (s *mySource) Parse(raw []byte) (dispatch.Parsed, error) {
+//	func (s *mySource) Parse(raw []byte) (dispatch.Message, error) {
 //	    var env struct {
 //	        Type    string          `json:"type"`
 //	        Payload json.RawMessage `json:"payload"`
 //	    }
 //	    if err := json.Unmarshal(raw, &env); err != nil {
-//	        return dispatch.Parsed{}, err
+//	        return dispatch.Message{}, err
 //	    }
-//	    if env.Type == "" {
-//	        return dispatch.Parsed{}, errors.New("missing type field")
-//	    }
-//	    return dispatch.Parsed{Key: env.Type, Payload: env.Payload}, nil
+//	    return dispatch.Message{Key: env.Type, Payload: env.Payload}, nil
 //	}
 type Source interface {
 	// Name returns the source identifier for logging and metrics.
@@ -84,39 +118,39 @@ type Source interface {
 	Discriminator() Discriminator
 
 	// Parse attempts to parse raw bytes as this source's format.
-	// Returns the parsed result and nil if successful, or an error describing
-	// why parsing failed.
-	Parse(raw []byte) (Parsed, error)
+	// Returns the parsed message and nil if successful, or an error
+	// describing why parsing failed.
+	Parse(raw []byte) (Message, error)
 }
 
 // SourceFunc creates a Source from a name, discriminator, and parse function.
-// Use this for simple sources that don't need a struct:
+// Use for simple sources that don't need a struct:
 //
 //	r.AddSource(dispatch.SourceFunc(
 //	    "legacy",
 //	    dispatch.HasFields("type", "payload"),
-//	    func(raw []byte) (dispatch.Parsed, error) {
+//	    func(raw []byte) (dispatch.Message, error) {
 //	        // parse logic
 //	    },
 //	))
-func SourceFunc(name string, disc Discriminator, parse func([]byte) (Parsed, error)) Source {
+func SourceFunc(name string, disc Discriminator, parse func([]byte) (Message, error)) Source {
 	return &sourceFunc{name: name, disc: disc, parse: parse}
 }
 
 type sourceFunc struct {
 	name  string
 	disc  Discriminator
-	parse func([]byte) (Parsed, error)
+	parse func([]byte) (Message, error)
 }
 
-func (s *sourceFunc) Name() string                     { return s.name }
-func (s *sourceFunc) Discriminator() Discriminator     { return s.disc }
-func (s *sourceFunc) Parse(raw []byte) (Parsed, error) { return s.parse(raw) }
+func (s *sourceFunc) Name() string                      { return s.name }
+func (s *sourceFunc) Discriminator() Discriminator      { return s.disc }
+func (s *sourceFunc) Parse(raw []byte) (Message, error) { return s.parse(raw) }
 
-// Parsed contains the result of source parsing.
-type Parsed struct {
+// Message contains the result of source parsing.
+type Message struct {
 	// Key is the routing key used to find the handler.
-	// This is matched against keys passed to Register.
+	// This is matched against keys passed to RegisterProc/RegisterFunc.
 	Key string
 
 	// Version is the schema version of the payload, if available.
@@ -126,12 +160,26 @@ type Parsed struct {
 	// Payload is the raw JSON to unmarshal into the handler's type.
 	Payload json.RawMessage
 
-	// Complete is called after the handler finishes, regardless of success or failure.
-	// Use this for transport-specific completion semantics like Step Functions
-	// SendTaskSuccess/SendTaskFailure.
+	// Replier handles sending responses back to the caller.
+	// For fire-and-forget sources (EventBridge, SNS), this is nil.
+	// For request-response sources (Step Functions), this sends results back.
+	//
+	// When Replier is set and a Func is registered:
+	//   - On success: router marshals result and calls Replier.Reply
+	//   - On error: router calls Replier.Fail
 	//
-	// If Complete is nil, no completion callback is made.
-	// If Complete returns an error, that error is returned from Process.
-	// The err parameter is the handler's error (or nil on success).
-	Complete func(ctx context.Context, err error) error
+	// When Replier is set and a Proc is registered:
+	//   - On success: router calls Replier.Reply with empty JSON ({})
+	//   - On error: router calls Replier.Fail
+	Replier Replier
+}
+
+// Replier sends responses back to the message originator.
+// Implement this for request-response transport patterns.
+type Replier interface {
+	// Reply sends a successful response with the given JSON payload.
+	Reply(ctx context.Context, result json.RawMessage) error
+
+	// Fail sends a failure response with the given error.
+	Fail(ctx context.Context, err error) error
 }