Database Connector Component

A complete example of a database component with connection pooling, modeled on the postgres_query component from database-module.

Overview

This component executes SQL SELECT queries against Postgres. It demonstrates:

• Process-wide connection pool caching keyed by DSN
• Parameterized queries
• A settings-gated error port with the canonical error shape
• Retryability marking for transient database failures

Complete Implementation

Shared Pool Package

Pools live outside the component so every invocation (and every component in the module) reuses the same connection pool per DSN:

// Package pool caches database connections across component invocations.
package pool

import (
    "context"
    "errors"
    "strings"
    "sync"

    "github.com/jackc/pgx/v5/pgconn"
    "github.com/jackc/pgx/v5/pgxpool"
)

var pgPools sync.Map // map[string]*pgxpool.Pool

// Postgres returns a cached pgx pool for the given DSN, creating one on first use.
func Postgres(ctx context.Context, dsn string) (*pgxpool.Pool, error) {
    if v, ok := pgPools.Load(dsn); ok {
        return v.(*pgxpool.Pool), nil
    }
    p, err := pgxpool.New(ctx, dsn)
    if err != nil {
        return nil, err
    }
    if actual, loaded := pgPools.LoadOrStore(dsn, p); loaded {
        p.Close()
        return actual.(*pgxpool.Pool), nil
    }
    return p, nil
}

// IsTransientPostgres reports whether a Postgres failure could clear on a
// backoff retry. A SQLSTATE reply means the server rejected the statement —
// permanent — except classes describing the server's condition: connection
// exception (08), insufficient resources (53), shutdown in progress
// (57P01-03), serialization/deadlock aborts (40001, 40P01). Anything that is
// not a server reply (dial failure, dropped socket) never ran — transient.
func IsTransientPostgres(err error) bool {
    var pgErr *pgconn.PgError
    if errors.As(err, &pgErr) {
        switch {
        case strings.HasPrefix(pgErr.Code, "08"), strings.HasPrefix(pgErr.Code, "53"),
            pgErr.Code == "57P01", pgErr.Code == "57P02", pgErr.Code == "57P03",
            pgErr.Code == "40001", pgErr.Code == "40P01":
            return true
        }
        return false
    }
    return true
}

Component

package postgresquery

import (
    "context"
    "fmt"

    "github.com/myorg/database-module/components/pool"
    "github.com/tiny-systems/module/api/v1alpha1"
    "github.com/tiny-systems/module/module"
    "github.com/tiny-systems/module/registry"
)

const (
    ComponentName = "postgres_query"
    RequestPort   = "request"
    ResponsePort  = "response"
    ErrorPort     = "error"
)

type Context any

// Row is a result row keyed by column name, so downstream edges can
// navigate $.rows[0].columnName.
type Row map[string]any

type Settings struct {
    EnableErrorPort bool `json:"enableErrorPort" required:"true" title:"Enable Error Port"`
}

type Request struct {
    Context Context `json:"context,omitempty" configurable:"true" title:"Context"`
    DSN     string  `json:"dsn" required:"true" title:"DSN" description:"Postgres connection string"`
    SQL     string  `json:"sql" required:"true" minLength:"1" title:"SQL" description:"SELECT with $1, $2, ... placeholders" format:"textarea"`
    Params  []any   `json:"params,omitempty" title:"Params"`
}

type Response struct {
    Context Context `json:"context,omitempty" configurable:"true" title:"Context"`
    Rows    []Row   `json:"rows" title:"Rows"`
    Count   int     `json:"count" title:"Count"`
}

type Component struct {
    module.Base
    settings Settings
}

func (c *Component) Instance() module.Component {
    return &Component{}
}

func (c *Component) GetInfo() module.ComponentInfo {
    return module.ComponentInfo{
        Name:        ComponentName,
        Description: "Postgres Query",
        Info:        "Runs SELECT against Postgres and returns rows as a list of objects keyed by column name. Connection pool is cached per DSN.",
        Tags:        []string{"Postgres", "SQL", "DB"},
    }
}

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

// Handle dispatches the RequestPort. System ports go through capabilities.
func (c *Component) Handle(ctx context.Context, handler module.Handler, port string, msg any) module.Result {
    if port != RequestPort {
        return module.Fail(fmt.Errorf("unknown port: %s", port))
    }
    in, ok := msg.(Request)
    if !ok {
        return module.Fail(fmt.Errorf("invalid request"))
    }
    return c.query(ctx, handler, in)
}

func (c *Component) query(ctx context.Context, handler module.Handler, in Request) module.Result {
    p, err := pool.Postgres(ctx, in.DSN)
    if err != nil {
        return c.fail(ctx, handler, in.Context, err)
    }

    rows, err := p.Query(ctx, in.SQL, in.Params...)
    if err != nil {
        return c.fail(ctx, handler, in.Context, c.retryable(err))
    }
    defer rows.Close()

    cols := rows.FieldDescriptions()
    out := make([]Row, 0)

    for rows.Next() {
        values, err := rows.Values()
        if err != nil {
            return c.fail(ctx, handler, in.Context, c.retryable(err))
        }
        row := make(Row, len(cols))
        for i, col := range cols {
            row[string(col.Name)] = values[i]
        }
        out = append(out, row)
    }
    if err := rows.Err(); err != nil {
        return c.fail(ctx, handler, in.Context, c.retryable(err))
    }

    return handler(ctx, ResponsePort, Response{
        Context: in.Context,
        Rows:    out,
        Count:   len(out),
    })
}

// retryable marks a SELECT failure the server or network could recover from.
// Safe here in a way it isn't for writes: this component only reads, so
// re-running the whole handler cannot double-apply anything. A SQL error
// (bad syntax, unknown column, permission denied) is left unmarked — the
// same statement would just fail again.
func (c *Component) retryable(err error) error {
    if pool.IsTransientPostgres(err) {
        return module.Retryable(err)
    }
    return err
}

func (c *Component) fail(ctx context.Context, handler module.Handler, reqCtx Context, err error) module.Result {
    if !c.settings.EnableErrorPort {
        // Bubble unchanged so retryability marked at the call site survives.
        return module.Fail(err)
    }
    // Canonical {context, error, retryable} payload
    return handler(ctx, ErrorPort, module.NewError(reqCtx, err))
}

func (c *Component) Ports() []module.Port {
    ports := []module.Port{
        {Name: v1alpha1.SettingsPort, Label: "Settings", Configuration: c.settings},
        {Name: RequestPort, Label: "Request", Configuration: Request{}, Position: module.Left},
        {Name: ResponsePort, Label: "Response", Source: true, Configuration: Response{}, Position: module.Right},
    }
    if !c.settings.EnableErrorPort {
        return ports
    }
    return append(ports, module.Port{
        Name: ErrorPort, Label: "Error", Source: true, Configuration: module.ErrorMessage{}, Position: module.Bottom,
    })
}

var (
    _ module.Component       = (*Component)(nil)
    _ module.SettingsHandler = (*Component)(nil)
)

func init() {
    registry.Register(&Component{})
}

For INSERT/UPDATE/DELETE, database-module ships a separate postgres_exec component — deliberately, because retry semantics differ: re-running a read is always safe, re-running a write is not, so exec must never blanket-mark failures retryable.

Usage Examples

Requests are configured on the incoming edge. Credentials come from secret placeholders, never literals:

SELECT Query

{
  "context": "{{$.context}}",
  "dsn": "postgres://app:[[secret:db/password]]@db.internal:5432/myapp",
  "sql": "SELECT id, name, email FROM users WHERE status = $1 LIMIT $2",
  "params": ["active", 100]
}

Parameter Mapping from Upstream Data

{
  "dsn": "postgres://app:[[secret:db/password]]@db.internal:5432/myapp",
  "sql": "SELECT * FROM orders WHERE user_id = $1",
  "params": ["{{$.context.userId}}"]
}

Response Examples

Query Result (on response)

{
  "context": { "userId": "42" },
  "rows": [
    { "id": 1, "name": "Alice", "email": "alice@example.com" },
    { "id": 2, "name": "Bob", "email": "bob@example.com" }
  ],
  "count": 2
}

Downstream edges navigate $.rows[0].name, $.count, and the passthrough $.context....

Error Result (on error, when enabled)

{
  "context": { "userId": "42" },
  "error": "FATAL: the database system is starting up (SQLSTATE 57P03)",
  "retryable": true
}

Key Patterns Demonstrated

1. Pool Caching Outside the Component

Component instances come and go; the sync.Map of pools lives for the process. LoadOrStore closes the loser of a racing create.

2. Retryability Is a Property of the Error

if pool.IsTransientPostgres(err) {
    return module.Retryable(err)
}

The classification lives where the knowledge lives (SQLSTATE class), the marking travels with the error through module.NewError, and the retry component downstream honors it. Unmarked errors are never retried.

3. Error Port Uses module.ErrorMessage

return handler(ctx, ErrorPort, module.NewError(reqCtx, err))

Using the SDK type (rather than a hand-rolled struct) guarantees the {context, error, retryable} shape the retry component and the platform expect.

4. Parameterized Queries Only

rows, err := p.Query(ctx, in.SQL, in.Params...)

Always $1, $2, ... placeholders with params — never interpolate user data into SQL strings on the edge.

Security Best Practices

1. Use Secret Placeholders for Credentials

"dsn": "postgres://app:[[secret:db/password]]@db.internal:5432/myapp"

2. Always Use Parameterized Queries

// GOOD
{ "sql": "SELECT * FROM users WHERE id = $1", "params": ["{{$.context.userId}}"] }

// BAD - SQL injection vulnerable!
{ "sql": "SELECT * FROM users WHERE id = {{$.context.userId}}" }

3. Read/Write Separation

Keep query (read) and exec (write) as separate components so retry policies can differ safely.

Extension Ideas

1. Row Shape Settings: let users declare the expected row schema in settings so downstream edges get typed autocompletion (the shipped postgres_query does this)
2. Transaction Support: a postgres_tx component scoping BEGIN/COMMIT around a subtree
3. Redis Components: the same pool pattern applies — see redis_get, redis_set, redis_dedup in database-module
4. Vector Search: vector_search / vector_upsert build on the same pool package for pgvector