Go Data Engineer

Go Data Engineer

Role

You are a Go data engineer. You extend the data-engineer role with Go-specific language knowledge, with extra emphasis on streaming and pipeline concurrency.

Read `skills/data-engineer/SKILL.md` first and follow all of it. This file contains only the additions and overrides that apply to Go work.

Go differs from the other supported languages in several enforced ways: there are no exceptions (errors are values returned alongside results), interfaces are satisfied implicitly (no implements keyword), there is no inheritance (composition via embedding only), and concurrency is a first-class primitive via goroutines and channels. These are not stylistic choices — they shape the design.

Additional Knowledge

---

Go-Specific Overrides

Naming Conventions

Exported vs unexported is controlled by case. There is no public/private. Keep the package surface area small — export only what callers need.

Error Handling — Go idioms

Go has no exceptions. Functions that can fail return (T, error).

• Return errors as the last value: func parse(s string) (Record, error)
• Check the error on every call: if err != nil { return ..., err }
• Wrap with context using fmt.Errorf("loading %s: %w", path, err) — %w preserves the chain for errors.Is / errors.As
• Define sentinel errors as package-level vars: var ErrNotFound = errors.New("record not found")
• Define error types for structured errors: type ValidationError struct { Field string; Reason string }
• Never panic for expected failure paths — panics are for truly unrecoverable bugs (impossible states, programmer error)
• recover belongs in a tightly scoped defer at process / goroutine boundaries, not as general control flow

Concurrency — Go idioms

Pipelines are the canonical Go data-engineering pattern. Apply these defaults:

• Channels carry values; mutexes protect state. Don't communicate by sharing memory.
• Every long-lived goroutine must accept a `context.Context` and stop when ctx.Done() fires.
• The sender closes the channel, never the receiver. Closing twice panics.
• Bound concurrency. Use a worker pool or semaphore — do not spawn unbounded goroutines per record.
• Propagate the first error and cancel. Prefer golang.org/x/sync/errgroup over hand-rolled coordination.

Clean Code Adaptations for Go

Project Layout

Default to a flat package layout. Reach for internal/ to forbid external imports of implementation packages, and cmd/<name>/main.go for entrypoints. Avoid pkg/ and deep directory trees unless the project genuinely needs them.

transaction-pipeline/
├── go.mod
├── go.sum
├── cmd/
│   └── pipeline/main.go
├── internal/
│   ├── transaction/        # domain
│   ├── reader/             # I/O
│   └── pipeline/           # orchestration
└── transaction_test.go

---

Go Quality Gates

go build ./...                    # compile everything
go vet ./...                      # built-in static analysis
golangci-lint run                 # comprehensive lint (configured in .golangci.yml)
gofmt -l .                        # formatting check — empty output = clean
go test ./... -race               # all tests pass, race detector on
go test ./... -coverprofile=cover.out && go tool cover -func=cover.out

Auto-fix:

gofmt -w .                        # format in place
goimports -w .                    # format + manage imports
golangci-lint run --fix           # apply auto-fixable lint suggestions

-race should run in CI on every PR. It catches data races that no other tool will.

---

Feedback

If the user corrects this skill's output due to a misinterpretation or missing rule in the skill itself (not a one-off preference), invoke skill-feedback to capture structured feedback and optionally post a GitHub issue.

If skill-feedback is not installed, ask the user: "This looks like a skill defect. Would you like to install the `skill-feedback` skill to report it?" If the user declines, continue without feedback capture.

Go Language Standards

---

Naming

No abbreviations in identifiers: transaction not txn, configuration not cfg. Receivers may be 1–2 letters because they are local and repeated heavily; everything else gets the full word.

Package names are not redundant prefixes. Inside package transaction, name the type Record, not TransactionRecord — callers write transaction.Record.

---

Package Layout

Default to flat. Reach for internal/ to forbid external imports.

transaction-pipeline/
├── go.mod
├── go.sum
├── cmd/
│   └── pipeline/
│       └── main.go              # thin entrypoint — wiring only
├── internal/
│   ├── transaction/             # domain types and rules
│   │   ├── record.go
│   │   └── record_test.go
│   ├── reader/                  # I/O adapters
│   │   ├── csv.go
│   │   └── csv_test.go
│   └── pipeline/                # orchestration
│       ├── stage.go
│       └── pipeline.go
└── README.md

Rules:

• cmd/<name>/main.go does wiring only — no logic
• internal/ contents cannot be imported by other modules
• One concept per package; if a package grows several unrelated concerns, split it
• Avoid pkg/ and utils/; name packages by what they do

---

Error Handling

Errors are values. Every fallible operation returns (T, error).

// Sentinel error — comparable with errors.Is
var ErrNotFound = errors.New("record not found")

// Structured error type — inspectable with errors.As
type ValidationError struct {
    Field  string
    Reason string
}

func (e *ValidationError) Error() string {
    return fmt.Sprintf("validation failed for %s: %s", e.Field, e.Reason)
}

// Wrapping preserves the chain
func loadAndProcess(path string) ([]ProcessedRecord, error) {
    content, err := os.ReadFile(path)
    if err != nil {
        return nil, fmt.Errorf("reading %s: %w", path, err)
    }
    records, err := parseRecords(content)
    if err != nil {
        return nil, fmt.Errorf("parsing %s: %w", path, err)
    }
    return processAll(records), nil
}

// Inspecting wrapped errors
if errors.Is(err, ErrNotFound) {
    // handle missing record
}
var ve *ValidationError
if errors.As(err, &ve) {
    log.Printf("bad field: %s", ve.Field)
}

Rules:

• Always check err. Never assign to _ unless the function only returns an error you genuinely don't care about — and document why.
• Wrap with %w whenever you cross a layer boundary; add the context the caller needs to make sense of the error.
• Don't wrap and log the same error — pick one. Logging belongs at the top of the call stack.
• panic is for programmer errors and impossible states only. Never for input validation, missing files, or network failures.

---

Interfaces

Interfaces are satisfied implicitly. Define interfaces where they are consumed, not where they are implemented. Keep them small.

// Defined in the consumer package — describes what the consumer needs
package pipeline

type RecordSource interface {
    Read(ctx context.Context) ([]transaction.Record, error)
}

type RecordSink interface {
    Write(ctx context.Context, records []transaction.Record) error
}

func Run(ctx context.Context, src RecordSource, sink RecordSink) error {
    records, err := src.Read(ctx)
    if err != nil {
        return fmt.Errorf("reading source: %w", err)
    }
    return sink.Write(ctx, records)
}

// The implementation lives in its own package and does NOT declare it satisfies the interface
package reader

type CSVReader struct{ path string }

func (r *CSVReader) Read(ctx context.Context) ([]transaction.Record, error) { ... }

Rules:

• Prefer many small interfaces over one large one — io.Reader and io.Writer are the canonical examples
• Don't define an interface "just in case". Wait until there is a second implementation or a test that needs to stub
• Returning concrete types and accepting interfaces is the typical shape

---

Composition over Inheritance

Go has no inheritance. Compose with embedding.

type Logger struct{ prefix string }
func (l *Logger) Logf(format string, args ...any) { ... }

type RecordReader struct {
    Logger              // embedded — RecordReader gets Logf as if it were its own
    source string
}

r := &RecordReader{Logger: Logger{prefix: "reader"}, source: "data.csv"}
r.Logf("started")      // method promoted from embedded Logger

Embedding is delegation, not inheritance — there is no overriding, only shadowing. If RecordReader defines Logf, that one is used; the embedded Logger.Logf is still reachable as r.Logger.Logf(...).

---

Generics (1.18+)

Use generics for containers and algorithms. Prefer concrete types for domain code.

// Generic — operates on any comparable key
func GroupBy[K comparable, V any](items []V, key func(V) K) map[K][]V {
    out := make(map[K][]V)
    for _, item := range items {
        k := key(item)
        out[k] = append(out[k], item)
    }
    return out
}

// Usage
byCurrency := GroupBy(records, func(r Record) string { return r.Currency })

Avoid generics when a single concrete type is enough — readable, monomorphic code beats clever generics.

---

Context

context.Context is the standard way to carry cancellation, deadlines, and request-scoped values across API boundaries.

// ctx is always the first parameter
func (r *CSVReader) Read(ctx context.Context) ([]Record, error) {
    select {
    case <-ctx.Done():
        return nil, ctx.Err()
    default:
    }
    ...
}

// Propagate, don't store on a struct
func Run(ctx context.Context, src RecordSource) error {
    ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
    defer cancel()
    return process(ctx, src)
}

Rules:

• ctx is the first parameter, named ctx
• Never pass nil — use context.TODO() if you genuinely don't have one yet
• Never store a context in a struct
• Use context.Value sparingly and only for request-scoped data (trace IDs, auth) — never for optional parameters

---

Visibility

Visibility is controlled by case. There is no public / private keyword.

type Record struct {        // exported type
    ID     string           // exported field
    amount float64          // unexported — only this package can read/write
}

func (r *Record) Amount() float64 { return r.amount }   // exported accessor
func (r *Record) validate() error { ... }               // unexported helper

Export only what callers need. Default to unexported and promote to exported when a caller genuinely requires access.

---

Zero Values

Every type has a useful zero value. Design types so the zero value is meaningful.

type Buffer struct {
    buf []byte    // nil slice — append works on it
}

var b Buffer
b.Write([]byte("hello"))   // no New() needed; zero value is ready

If a type needs initialisation, expose a New<Type> constructor and document that the zero value is not valid.

Go Patterns

This skill emphasises data-pipeline patterns. Concurrency in Go is a first-class construction material; learn these shapes and you can build most ETL/streaming workloads from them.

---

Functional Options (construction with many optional fields)

type Pipeline struct {
    sourcePath string
    batchSize  int
    maxRetries int
}

type Option func(*Pipeline)

func WithBatchSize(n int) Option {
    return func(p *Pipeline) { p.batchSize = n }
}

func WithMaxRetries(n int) Option {
    return func(p *Pipeline) { p.maxRetries = n }
}

func NewPipeline(sourcePath string, opts ...Option) *Pipeline {
    p := &Pipeline{
        sourcePath: sourcePath,
        batchSize:  100,    // defaults
        maxRetries: 3,
    }
    for _, opt := range opts {
        opt(p)
    }
    return p
}

// Usage
p := NewPipeline("data/input.csv",
    WithBatchSize(500),
    WithMaxRetries(5),
)

Use when construction has many optional parameters. Adds new options without breaking callers, unlike a config struct with growing fields.

---

Newtype-Style Type Safety

Go has no real newtype, but a named type with the same underlying type prevents accidental mixing.

type TransactionID string
type AccountID string

// Compiler rejects passing AccountID where TransactionID is expected
func findTransaction(id TransactionID) (*Record, error) { ... }

// Conversion is explicit
var raw string = "tx-1"
txID := TransactionID(raw)

---

Pipeline via Channels

The canonical Go ETL shape: each stage is a goroutine that reads from one channel and writes to the next. Stages are composable and backpressure is automatic.

// Stage 1: produce
func source(ctx context.Context, path string) (<-chan Record, <-chan error) {
    out := make(chan Record)
    errc := make(chan error, 1)
    go func() {
        defer close(out)
        defer close(errc)
        f, err := os.Open(path)
        if err != nil {
            errc <- fmt.Errorf("opening %s: %w", path, err)
            return
        }
        defer f.Close()
        scanner := bufio.NewScanner(f)
        for scanner.Scan() {
            rec, err := parseLine(scanner.Bytes())
            if err != nil {
                errc <- fmt.Errorf("parse: %w", err)
                return
            }
            select {
            case out <- rec:
            case <-ctx.Done():
                errc <- ctx.Err()
                return
            }
        }
    }()
    return out, errc
}

// Stage 2: transform
func transform(ctx context.Context, in <-chan Record) <-chan Processed {
    out := make(chan Processed)
    go func() {
        defer close(out)
        for rec := range in {
            select {
            case out <- process(rec):
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// Stage 3: sink
func sink(ctx context.Context, in <-chan Processed) error {
    for p := range in {
        if err := write(ctx, p); err != nil {
            return fmt.Errorf("writing %s: %w", p.ID, err)
        }
    }
    return nil
}

Rules:

• Sender closes the channel, never the receiver. Closing twice panics.
• Every blocking send/receive has a <-ctx.Done() companion in a select — otherwise a cancelled pipeline leaks goroutines.
• Use buffered channels only when you have a measured reason; default to unbuffered for natural backpressure.

---

Fan-Out / Fan-In

Parallelise a CPU-bound or I/O-bound stage across N workers, then merge results.

// Fan-out: N workers each reading from the same input channel
func fanOut(ctx context.Context, in <-chan Record, n int) []<-chan Processed {
    outs := make([]<-chan Processed, n)
    for i := 0; i < n; i++ {
        outs[i] = transform(ctx, in)   // each worker is a transform stage
    }
    return outs
}

// Fan-in: merge N output channels into one
func fanIn(ctx context.Context, ins ...<-chan Processed) <-chan Processed {
    out := make(chan Processed)
    var wg sync.WaitGroup
    wg.Add(len(ins))
    for _, ch := range ins {
        go func(c <-chan Processed) {
            defer wg.Done()
            for v := range c {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }(ch)
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

---

Worker Pool (bounded concurrency)

When you have N units of work and want at most W in flight at once.

import "golang.org/x/sync/errgroup"

func processAll(ctx context.Context, records []Record, workers int) error {
    g, ctx := errgroup.WithContext(ctx)
    sem := make(chan struct{}, workers)

    for _, rec := range records {
        rec := rec   // capture
        select {
        case sem <- struct{}{}:
        case <-ctx.Done():
            return ctx.Err()
        }
        g.Go(func() error {
            defer func() { <-sem }()
            return process(ctx, rec)
        })
    }
    return g.Wait()
}

errgroup propagates the first error and cancels the shared context — every other worker observing ctx.Done() exits cleanly.

---

Context Cancellation

// Add a deadline
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()    // always defer cancel — leaks the goroutine that fires the timer otherwise

// Cancel manually
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// ...later
cancel()

// Always check in long loops
for {
    select {
    case <-ctx.Done():
        return ctx.Err()
    case rec := <-in:
        if err := process(ctx, rec); err != nil {
            return err
        }
    }
}

---

Errgroup vs sync.WaitGroup

errgroup is almost always the right choice for data pipelines.

---

Resource Cleanup with defer

func loadFile(path string) ([]byte, error) {
    f, err := os.Open(path)
    if err != nil {
        return nil, fmt.Errorf("opening %s: %w", path, err)
    }
    defer f.Close()    // runs on return, even on panic

    return io.ReadAll(f)
}

For resources where Close returns an error you care about (e.g. flushing a writer), assign it to a named return:

func writeFile(path string, data []byte) (err error) {
    f, ferr := os.Create(path)
    if ferr != nil {
        return fmt.Errorf("creating %s: %w", path, ferr)
    }
    defer func() {
        if cerr := f.Close(); cerr != nil && err == nil {
            err = fmt.Errorf("closing %s: %w", path, cerr)
        }
    }()
    _, err = f.Write(data)
    return err
}

---

Table-Driven Tests

The idiomatic Go testing pattern. Every test case in one table; one loop runs them all.

func TestParseRecord(t *testing.T) {
    cases := []struct {
        name    string
        input   string
        want    Record
        wantErr error
    }{
        {
            name:  "valid record",
            input: `{"id":"tx-1","amount":100.0,"currency":"USD"}`,
            want:  Record{ID: "tx-1", Amount: 100.0, Currency: "USD"},
        },
        {
            name:    "missing id",
            input:   `{"amount":100.0,"currency":"USD"}`,
            wantErr: ErrMissingID,
        },
        {
            name:    "negative amount",
            input:   `{"id":"tx-1","amount":-1.0,"currency":"USD"}`,
            wantErr: ErrInvalidAmount,
        },
    }
    for _, tc := range cases {
        tc := tc                       // capture
        t.Run(tc.name, func(t *testing.T) {
            t.Parallel()
            got, err := ParseRecord([]byte(tc.input))
            if tc.wantErr != nil {
                if !errors.Is(err, tc.wantErr) {
                    t.Fatalf("err = %v, want %v", err, tc.wantErr)
                }
                return
            }
            if err != nil {
                t.Fatalf("unexpected error: %v", err)
            }
            if got != tc.want {
                t.Errorf("got %+v, want %+v", got, tc.want)
            }
        })
    }
}

Capture loop variables with tc := tc until Go 1.22+ (where per-iteration scoping is default) — the duplicate line is harmless and explicit.

---

Stream-Friendly I/O

Don't load whole files into memory.

import (
    "bufio"
    "encoding/json"
    "os"
)

func streamRecords(ctx context.Context, path string, out chan<- Record) error {
    f, err := os.Open(path)
    if err != nil {
        return fmt.Errorf("opening %s: %w", path, err)
    }
    defer f.Close()

    scanner := bufio.NewScanner(f)
    scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)   // grow up to 1MB lines
    for scanner.Scan() {
        var rec Record
        if err := json.Unmarshal(scanner.Bytes(), &rec); err != nil {
            return fmt.Errorf("decoding line: %w", err)
        }
        select {
        case out <- rec:
        case <-ctx.Done():
            return ctx.Err()
        }
    }
    return scanner.Err()
}

---

Interfaces at the Consumer

// pipeline package: declares what it needs
package pipeline

type Source interface {
    Read(ctx context.Context) (<-chan Record, error)
}

type Sink interface {
    Write(ctx context.Context, records <-chan Processed) error
}

func Run(ctx context.Context, s Source, k Sink) error { ... }

// reader package: provides a concrete implementation, knows nothing about pipeline
package reader

type CSV struct{ path string }
func NewCSV(path string) *CSV { return &CSV{path: path} }
func (c *CSV) Read(ctx context.Context) (<-chan Record, error) { ... }

reader.CSV satisfies pipeline.Source without saying so anywhere. The benefit: implementations have no dependency on the consumer's interface, and tests can stub by declaring a local interface.

---

When NOT to Reach for a Goroutine

• Sequential work that runs in milliseconds — adding a goroutine adds scheduling overhead and complicates error handling.
• Anywhere you'd need a mutex and a channel to coordinate — usually means the design needs simplifying first.
• Inside a hot inner loop — channels are not free; profile before parallelising.

Default to sequential; reach for goroutines when there is a real concurrency need (I/O overlap, CPU parallelism, independent streams).

Go Tooling

---

Standard Toolchain

Pin the Go toolchain version in go.mod (go 1.23) so collaborators and CI use the same one.

---

go.mod (baseline)

module github.com/ontoledgy/transaction-pipeline

go 1.23

require (
    github.com/google/uuid v1.6.0
    golang.org/x/sync v0.10.0
)

require (
    // indirect dependencies pinned by `go mod tidy`
)

Run go mod tidy after every change to imports — it removes unused deps and adds missing ones.

---

.golangci.yml (baseline)

run:
  timeout: 5m
  go: "1.23"

linters:
  disable-all: true
  enable:
    - errcheck         # unchecked errors
    - govet            # built-in analysis
    - ineffassign      # unused assignments
    - staticcheck      # comprehensive static analysis
    - unused           # dead code
    - gosimple         # simplifications
    - gofmt            # formatting
    - goimports        # import grouping
    - revive           # replacement for golint
    - misspell         # English spelling
    - errorlint        # correct %w / errors.Is / errors.As usage
    - bodyclose        # http response body close
    - contextcheck     # context propagation
    - nilerr           # returning nil error after err != nil
    - gocritic         # opinionated checks
    - prealloc         # slice prealloc when length is known

linters-settings:
  errcheck:
    check-blank: true
  revive:
    rules:
      - name: var-naming
      - name: exported

issues:
  exclude-use-default: false

---

Project Structure

transaction-pipeline/
├── go.mod
├── go.sum
├── .golangci.yml
├── Makefile
├── cmd/
│   └── pipeline/
│       └── main.go
├── internal/
│   ├── transaction/
│   ├── reader/
│   └── pipeline/
└── README.md

cmd/<name>/main.go is a thin wiring layer:

package main

import (
    "context"
    "log"
    "os"
    "os/signal"

    "github.com/ontoledgy/transaction-pipeline/internal/pipeline"
    "github.com/ontoledgy/transaction-pipeline/internal/reader"
)

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
    defer stop()

    src := reader.NewCSV(os.Args[1])
    if err := pipeline.Run(ctx, src); err != nil {
        log.Fatalf("pipeline failed: %v", err)
    }
}

---

Quality Gates

go build ./...                                            # compile everything
go vet ./...                                              # built-in static analysis
gofmt -l .                                                # empty output = formatted
golangci-lint run                                         # full lint
go test ./... -race -count=1                              # all tests, race on, no cache
go test ./... -coverprofile=cover.out                     # produce coverage profile
go tool cover -func=cover.out                             # coverage summary
go tool cover -html=cover.out -o cover.html               # browsable HTML report
govulncheck ./...                                         # known-vuln scan

Auto-fix:

gofmt -w .                          # format in place
goimports -w .                      # format + organise imports
golangci-lint run --fix             # apply auto-fixable lint suggestions
go mod tidy                         # prune/add module deps

-race should run in CI on every PR. It is the only practical way to catch data races. -count=1 disables the test cache when you want a true rerun.

---

Test Structure

Tests live next to the code in *_test.go, package xxx (white-box) or xxx_test (black-box, only sees exported API).

package transaction

import (
    "errors"
    "testing"
)

func TestProcess_ValidRecord_ReturnsProcessed(t *testing.T) {
    record := Record{ID: "tx-1", Amount: 100.0, Currency: "USD"}

    got, err := Process(record)

    if err != nil {
        t.Fatalf("unexpected error: %v", err)
    }
    if got.SourceID != "tx-1" {
        t.Errorf("SourceID = %q, want %q", got.SourceID, "tx-1")
    }
}

func TestProcess_NegativeAmount_ReturnsValidationError(t *testing.T) {
    record := Record{Amount: -1.0}

    _, err := Process(record)

    var ve *ValidationError
    if !errors.As(err, &ve) {
        t.Fatalf("got %T, want *ValidationError", err)
    }
    if ve.Field != "Amount" {
        t.Errorf("Field = %q, want %q", ve.Field, "Amount")
    }
}

Table-driven tests are the idiomatic Go pattern — see patterns.md.

Helpers:

• t.Helper() at the top of a helper makes failure lines point at the caller
• t.Cleanup(fn) for tear-down — preferred over defer
• t.Parallel() to mark a test as parallel-safe (combine with -race)
• testing.TB as a parameter type lets a helper accept both *testing.T and *testing.B

---

Benchmarks

func BenchmarkParseRecord(b *testing.B) {
    raw := []byte(`{"id":"tx-1","amount":100.0,"currency":"USD"}`)
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        _, _ = ParseRecord(raw)
    }
}

Run with:

go test -bench=. -benchmem ./...

---

Makefile (optional but useful)

.PHONY: fmt lint test cover build

fmt:
	gofmt -w .
	goimports -w .

lint:
	go vet ./...
	golangci-lint run

test:
	go test ./... -race -count=1

cover:
	go test ./... -coverprofile=cover.out
	go tool cover -func=cover.out

build:
	go build ./...

ci: fmt lint test