Component Patterns

This guide covers common patterns for building TinySystems components. These patterns come from production modules (common-module, http-module, kubernetes-module).

Pattern Categories

• Processing Patterns - Transform, filter, route data
• Source Patterns - Generate data, respond to events
• Sink Patterns - Send data to external systems
• Stateful Patterns - Manage component state

Throughout: Handle returns module.Result, settings arrive via OnSettings, control clicks via OnControl, and reconciles via OnReconcile. System ports never reach Handle.

Processing Patterns

Transformer

Converts input to output format. This is the real transform (modify) component:

type Context any

type InMessage struct {
    Context Context `json:"context" configurable:"true" required:"true" title:"Context"`
}

type OutMessage struct {
    Context Context `json:"context" configurable:"true" title:"Context"`
}

type Component struct{}

func (t *Component) Handle(ctx context.Context, handler module.Handler, port string, msg interface{}) module.Result {
    if in, ok := msg.(InMessage); ok {
        return handler(ctx, OutPort, OutMessage{Context: in.Context})
    }
    return module.Fail(fmt.Errorf("invalid message"))
}

func (t *Component) Ports() []module.Port {
    return []module.Port{
        {Name: InPort, Label: "In", Position: module.Left, Configuration: InMessage{}},
        {Name: OutPort, Label: "Out", Source: true, Position: module.Right, Configuration: new(OutMessage)},
    }
}

Keep the passthrough payload under a context key so a downstream edge reads $.context.<field> — the mid-chain convention.

Filter

Routes messages based on a condition, with an error port:

type Component struct {
    settings Settings
}

type Settings struct {
    Condition string `json:"condition" title:"Filter Expression" required:"true"`
}

func (f *Component) OnSettings(_ context.Context, msg any) error {
    in, ok := msg.(Settings)
    if !ok {
        return fmt.Errorf("invalid settings")
    }
    f.settings = in
    return nil
}

func (f *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    in, ok := msg.(Message)
    if !ok {
        return module.Fail(fmt.Errorf("invalid message"))
    }

    match, err := f.evaluate(in.Data, f.settings.Condition)
    if err != nil {
        // Canonical error-port payload: {context, error, retryable}
        return output(ctx, "error", module.NewError(in, err))
    }

    if match {
        return output(ctx, "match", in)
    }
    return output(ctx, "no_match", in)
}

func (f *Component) Ports() []module.Port {
    return []module.Port{
        {Name: v1alpha1.SettingsPort, Label: "Settings", Configuration: f.settings},
        {Name: "input", Label: "Input", Position: module.Left, Configuration: Message{}},
        {Name: "match", Label: "Match", Source: true, Position: module.Right, Configuration: new(Message)},
        {Name: "no_match", Label: "No Match", Source: true, Position: module.Bottom, Configuration: new(Message)},
        {Name: "error", Label: "Error", Source: true, Position: module.Bottom, Configuration: module.ErrorMessage{}},
    }
}

Router

Routes to dynamically generated outputs. Condensed from the real router component — routes are settings, output ports are generated per route, and the message carries per-route boolean conditions evaluated on the edge:

type Settings struct {
    Routes            []string `json:"routes" required:"true" title:"Routes" minItems:"1" uniqueItems:"true"`
    EnableDefaultPort bool     `json:"enableDefaultPort" required:"true" title:"Enable default port"`
}

type Condition struct {
    RouteName string `json:"route" title:"Route" required:"true"`
    Condition bool   `json:"condition" required:"true" title:"Condition"`
}

type InMessage struct {
    Context    Context     `json:"context" configurable:"true" required:"true" title:"Context"`
    Conditions []Condition `json:"conditions" required:"true" title:"Conditions" minItems:"1" uniqueItems:"true"`
}

type OutMessage struct {
    Context Context `json:"context" configurable:"true" title:"Context"`
}

func (t *Component) Handle(ctx context.Context, handler module.Handler, port string, msg interface{}) module.Result {
    in, ok := msg.(InMessage)
    if !ok {
        return module.Fail(fmt.Errorf("invalid message on port %q", port))
    }

    for _, condition := range in.Conditions {
        if !condition.Condition {
            continue
        }
        if t.hasRoute(condition.RouteName) {
            return handler(ctx, getPortNameFromRoute(condition.RouteName), OutMessage{Context: in.Context})
        }
        break
    }
    if t.settings.EnableDefaultPort {
        return handler(ctx, DefaultPort, OutMessage{Context: in.Context})
    }
    return module.Fail(fmt.Errorf("no matching route: %v", in.Conditions))
}

func (t *Component) Ports() []module.Port {
    ports := []module.Port{
        {Name: v1alpha1.SettingsPort, Label: "Settings", Configuration: t.settings},
        {Name: InPort, Label: "IN", Position: module.Left, Configuration: t.inMessageSample()},
    }
    for _, r := range t.settings.Routes {
        ports = append(ports, module.Port{
            Name:          getPortNameFromRoute(r), // "out_" + lowercase(route)
            Label:         r,
            Source:        true,
            Position:      module.Right,
            Configuration: new(OutMessage),
        })
    }
    if t.settings.EnableDefaultPort {
        ports = append(ports, module.Port{
            Name: DefaultPort, Label: "Default", Source: true,
            Position: module.Bottom, Configuration: new(OutMessage),
        })
    }
    return ports
}

Splitter

Splits arrays into individual items, respecting cancellation:

type Input struct {
    Items []any `json:"items" title:"Items" required:"true"`
}

type Output struct {
    Item  any `json:"item"`
    Index int `json:"index"`
    Total int `json:"total"`
}

func (s *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    in, ok := msg.(Input)
    if !ok {
        return module.Fail(fmt.Errorf("invalid message"))
    }
    total := len(in.Items)

    for i, item := range in.Items {
        select {
        case <-ctx.Done():
            return module.Fail(ctx.Err())
        default:
            if res := output(ctx, "item", Output{Item: item, Index: i, Total: total}); res.IsErr() {
                return res
            }
        }
    }
    return output(ctx, "done", struct {
        Total int `json:"total"`
    }{Total: total})
}

Aggregator

Collects items and emits batches:

type Component struct {
    settings Settings
    items    []any
    mu       sync.Mutex
}

type Settings struct {
    BatchSize int `json:"batchSize" title:"Batch Size" default:"10" minimum:"1"`
}

func (a *Component) OnSettings(_ context.Context, msg any) error {
    in, ok := msg.(Settings)
    if !ok {
        return fmt.Errorf("invalid settings")
    }
    a.settings = in
    return nil
}

func (a *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    switch port {
    case "item":
        a.mu.Lock()
        a.items = append(a.items, msg)
        var batch []any
        if len(a.items) >= a.settings.BatchSize {
            batch = a.items
            a.items = nil
        }
        a.mu.Unlock()

        if batch != nil {
            return output(ctx, "batch", batch)
        }
        return module.Ok(nil)

    case "flush":
        a.mu.Lock()
        batch := a.items
        a.items = nil
        a.mu.Unlock()

        if len(batch) > 0 {
            return output(ctx, "batch", batch)
        }
        return module.Ok(nil)
    }
    return module.Fail(fmt.Errorf("unknown port %q", port))
}

Source Patterns

Ticker

Emits messages at intervals. Source components have no business inputs — everything happens in OnControl, OnReconcile, and a background loop emitting via module.Base:

type Component struct {
    module.Base

    settings       Settings
    cancelFunc     context.CancelFunc
    cancelFuncLock *sync.Mutex
}

func (t *Component) OnControl(ctx context.Context, msg any) error {
    if msg == nil || !utils.IsLeader(ctx) {
        return nil
    }
    switch ctrl := msg.(type) {
    case ControlRunning:
        if ctrl.Stop {
            t.clearMetadata()
            return t.stop()
        }
    case ControlStopped:
        t.settings.Context = ctrl.Context
        t.persistMetadata()
        if t.isRunning() {
            return nil
        }
        go t.run(context.Background())
    }
    return nil
}

// OnReconcile restores the running loop after a pod restart.
func (t *Component) OnReconcile(ctx context.Context, node v1alpha1.TinyNode) error {
    if node.Status.Metadata == nil {
        return nil
    }
    if _, running := node.Status.Metadata["ticker-running"]; !running {
        return nil
    }
    if t.isRunning() || !utils.IsLeader(ctx) {
        return nil
    }
    go t.run(ctx)
    return nil
}

func (t *Component) run(ctx context.Context) {
    runCtx, cancel := context.WithCancel(context.Background())
    defer cancel()
    t.setCancelFunc(cancel)

    t.Emit(context.Background(), v1alpha1.ControlPort, t.getControl()) // widget -> Running
    defer func() {
        t.setCancelFunc(nil)
        t.Emit(context.Background(), v1alpha1.ControlPort, t.getControl()) // widget -> Stopped
    }()

    timer := time.NewTimer(0)
    defer timer.Stop()
    for {
        select {
        case <-timer.C:
            if err := t.Emit(runCtx, OutPort, t.settings.Context).Err(); err != nil {
                // log downstream error, keep ticking
            }
            timer.Reset(time.Duration(t.settings.Delay) * time.Millisecond)
        case <-runCtx.Done():
            return
        }
    }
}

// persistMetadata survives pod restarts: emit a node-updater on _reconcile
func (t *Component) persistMetadata() {
    configBytes, _ := json.Marshal(t.settings)
    t.Emit(context.Background(), v1alpha1.ReconcilePort, func(n *v1alpha1.TinyNode) error {
        if n.Status.Metadata == nil {
            n.Status.Metadata = make(map[string]string)
        }
        n.Status.Metadata["ticker-running"] = "true"
        n.Status.Metadata["ticker-config"] = string(configBytes)
        return nil
    })
}

See Control Ports for the full ticker.

HTTP Server

Receives HTTP requests and forwards them into the flow as a blocking request/reply. Key structural points from the real http-module server:

func (h *Component) Ports() []module.Port {
    return []module.Port{
        {Name: v1alpha1.ClientPort},    // K8s client injection (ClientAware)
        {Name: v1alpha1.ReconcilePort}, // reconcile participation
        {Name: v1alpha1.SettingsPort, Label: "Settings", Configuration: h.settings},
        {
            Name:                  RequestPort,
            Label:                 "Request",
            Source:                true,
            Position:              module.Right,
            Configuration:         Request{},
            ResponseConfiguration: Response{}, // synchronous reply shape
        },
        {
            Name:          ResponsePort,
            Label:         "Response",
            Position:      module.Right,
            Configuration: Response{StatusCode: 200},
        },
        {Name: v1alpha1.ControlPort, Label: "Dashboard", Source: true, Configuration: h.getControl()},
    }
}

// OnClient receives the K8s client wrapper (for Service/Ingress management).
func (h *Component) OnClient(k8sClient module.K8sClient) {
    h.portManager = newPortManager(k8sClient)
}

Inside the HTTP handler, the server emits on its Request port and blocks on the Result that flows back from downstream — this is why chaining Result returns matters:

res := h.Emit(reqCtx, RequestPort, request)
if err := res.Err(); err != nil {
    // respond 5xx
}
// res.Value() carries the Response produced downstream

Signal (Manual Trigger)

Sends a message on button click — fire-and-forget, no running state:

type Control struct {
    Context Context `json:"context" required:"true" title:"Context"`
    Send    bool    `json:"send" format:"button" title:"Send" required:"true"`
}

func (t *Component) OnControl(ctx context.Context, msg any) error {
    if !utils.IsLeader(ctx) {
        return nil
    }
    ctrl, ok := msg.(Control)
    if !ok {
        return fmt.Errorf("invalid control msg: expected Control, got %T", msg)
    }
    if !ctrl.Send {
        return nil
    }

    sendCtx := ctrl.Context
    if sendCtx == nil {
        sendCtx = t.settings.Context
    }
    go t.Emit(context.Background(), OutPort, sendCtx)
    return nil
}

Sink Patterns

HTTP Client

Sends HTTP requests; marks transient failures retryable and uses the canonical error port:

type Component struct {
    settings Settings
    client   *http.Client
}

func (c *Component) OnSettings(_ context.Context, msg any) error {
    in, ok := msg.(Settings)
    if !ok {
        return fmt.Errorf("invalid settings")
    }
    c.settings = in
    c.client = &http.Client{
        Timeout: time.Duration(in.Timeout) * time.Millisecond,
    }
    return nil
}

func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    req, ok := msg.(Request)
    if !ok {
        return module.Fail(fmt.Errorf("invalid message"))
    }

    resp, err := c.doRequest(ctx, req)
    if err != nil {
        // NewError derives `retryable` from the error marking
        return output(ctx, "error", module.NewError(req, err))
    }
    return output(ctx, "response", resp)
}

func (c *Component) doRequest(ctx context.Context, req Request) (*Response, error) {
    httpReq, err := http.NewRequestWithContext(ctx, req.Method, c.settings.BaseURL+req.Path, bodyOf(req))
    if err != nil {
        return nil, module.Permanent(err) // bad input — never retry
    }

    resp, err := c.client.Do(httpReq)
    if err != nil {
        return nil, module.Retryable(err) // network error — retry can clear it
    }
    defer resp.Body.Close()

    respBody, _ := io.ReadAll(resp.Body)
    if resp.StatusCode >= 500 || resp.StatusCode == 429 {
        return nil, module.Retryable(fmt.Errorf("upstream %d: %s", resp.StatusCode, respBody))
    }
    if resp.StatusCode >= 400 {
        return nil, module.Permanent(fmt.Errorf("upstream %d: %s", resp.StatusCode, respBody))
    }
    return &Response{StatusCode: resp.StatusCode, Body: string(respBody)}, nil
}

func (c *Component) Ports() []module.Port {
    return []module.Port{
        {Name: v1alpha1.SettingsPort, Label: "Settings", Configuration: c.settings},
        {Name: "request", Label: "Request", Position: module.Left, Configuration: Request{}},
        {Name: "response", Label: "Response", Source: true, Position: module.Right, Configuration: new(Response)},
        {Name: "error", Label: "Error", Source: true, Position: module.Bottom, Configuration: module.ErrorMessage{}},
    }
}

Debug (Logger)

Logs messages and passes them through:

func (d *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    in, ok := msg.(Input)
    if !ok {
        return module.Fail(fmt.Errorf("invalid message"))
    }
    log.Info().Str("label", d.settings.Label).Interface("data", in.Data).Msg("debug")
    return output(ctx, "output", in) // pass through
}

Stateful Patterns

In-Memory State

Fine for state that may be lost on restart:

type Component struct {
    count int64
    mu    sync.Mutex
}

func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    switch port {
    case "increment":
        c.mu.Lock()
        c.count++
        count := c.count
        c.mu.Unlock()
        return output(ctx, "output", map[string]int64{"count": count})

    case "reset":
        c.mu.Lock()
        c.count = 0
        c.mu.Unlock()
        return output(ctx, "output", map[string]int64{"count": 0})
    }
    return module.Fail(fmt.Errorf("unknown port %q", port))
}

Durable State via the State Backend

For state that must survive restarts and be visible across replicas, embed module.Base and use the injected State (values are opaque []byte; serialize as needed):

type Component struct {
    module.Base
}

func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    st := c.State()
    if st == nil {
        return module.Fail(fmt.Errorf("state backend not available"))
    }

    // Read
    raw, ok, err := st.Get(ctx, "counter")
    if err != nil {
        return module.Fail(err)
    }
    count := int64(0)
    if ok {
        count, _ = strconv.ParseInt(string(raw), 10, 64)
    }

    // Write (debounced patch via the reconcile machinery)
    count++
    if err := st.Set(ctx, "counter", []byte(strconv.FormatInt(count, 10))); err != nil {
        return module.Fail(err)
    }
    return output(ctx, "output", map[string]int64{"count": count})
}

For run-scoped durable state shared across pods, use st.Scoped(module.ScopeExecution, runID).

Best Practices Summary

Next Steps

• Component Interface - Interface details
• System Ports - System port reference
• Multi-Replica Coordination - Scaling patterns