Template Method Pattern #

// DataProcessor defines the steps that can vary.
type DataProcessor interface {
    Validate() error
    FetchData() ([]byte, error)
    Format(data []byte) ([]byte, error)
    Save(formatted []byte) error
}

// Run is the template method — it defines the algorithm sequence.
// This function is Go's version of the "abstract class".
func Run(processor DataProcessor) error {
    // Fixed step: validate first
    if err := processor.Validate(); err != nil {
        return fmt.Errorf("validation failed: %w", err)
    }

    // Variable step: fetch the data
    data, err := processor.FetchData()
    if err != nil {
        return fmt.Errorf("fetch failed: %w", err)
    }

    // Fixed step: log progress (the same for all)
    log.Printf("data fetched: %d bytes", len(data))

    // Variable step: format
    formatted, err := processor.Format(data)
    if err != nil {
        return fmt.Errorf("format failed: %w", err)
    }

    // Variable step: save
    if err := processor.Save(formatted); err != nil {
        return fmt.Errorf("save failed: %w", err)
    }

    return nil
}

// Usage:
Run(&CSVProcessor{...})
Run(&JSONProcessor{...})

// BaseProcessor provides default implementations for hook methods
// and holds the template method.
type BaseProcessor struct {
    impl ProcessorImpl // interface to the concrete implementation
}

type ProcessorImpl interface {
    Validate() error
    FetchData() ([]byte, error)
    Format(data []byte) ([]byte, error)
    Save(formatted []byte) error
    // Hook methods — default implementations available
    BeforeFormat(data []byte) []byte   // default: return data unchanged
    AfterSave() error                  // default: no-op
}

// Run is the template method that cannot be overridden.
func (b *BaseProcessor) Run() error {
    if err := b.impl.Validate(); err != nil {
        return fmt.Errorf("validation failed: %w", err)
    }
    data, err := b.impl.FetchData()
    if err != nil {
        return fmt.Errorf("fetch failed: %w", err)
    }
    data = b.impl.BeforeFormat(data) // hook — optional
    formatted, err := b.impl.Format(data)
    if err != nil {
        return fmt.Errorf("format failed: %w", err)
    }
    if err := b.impl.Save(formatted); err != nil {
        return fmt.Errorf("save failed: %w", err)
    }
    return b.impl.AfterSave() // hook — optional
}

package pipeline

import (
    "context"
    "fmt"
    "log/slog"
    "time"
)

// PipelineResult stores a summary of the pipeline execution.
type PipelineResult struct {
    ProcessorName string
    RecordsRead   int
    RecordsWritten int
    Duration      time.Duration
    Error         error
}

// DataPipeline defines the steps every concrete pipeline must implement.
type DataPipeline interface {
    // Name returns the pipeline name for logging.
    Name() string

    // Validate validates the configuration and resource availability.
    Validate(ctx context.Context) error

    // Extract fetches data from the source.
    Extract(ctx context.Context) ([]Record, error)

    // Transform changes the data (validation, normalization, conversion).
    // Default implementation: return records unchanged (hook).
    Transform(ctx context.Context, records []Record) ([]Record, error)

    // Load stores the data to the destination.
    Load(ctx context.Context, records []Record) (int, error)

    // OnSuccess is called after the pipeline succeeds (hook — optional).
    OnSuccess(ctx context.Context, result PipelineResult)

    // OnError is called if the pipeline fails (hook — optional).
    OnError(ctx context.Context, err error)
}

// Record represents one row of data flowing through the pipeline.
type Record map[string]interface{}

// BasePipeline provides default implementations for the hook methods.
// All concrete pipelines may embed this struct to avoid
// duplicating hook implementations they do not need.
type BasePipeline struct{}

// Default Transform: return records unchanged.
func (b *BasePipeline) Transform(ctx context.Context, records []Record) ([]Record, error) {
    return records, nil
}

// Default OnSuccess: no-op.
func (b *BasePipeline) OnSuccess(ctx context.Context, result PipelineResult) {}

// Default OnError: no-op.
func (b *BasePipeline) OnError(ctx context.Context, err error) {}

// Execute is the Template Method — it defines the pipeline step order.
// This order CANNOT be changed by concrete implementations.
// Every concrete pipeline gets the same, guaranteed sequence.
func Execute(ctx context.Context, pipeline DataPipeline, logger *slog.Logger) PipelineResult {
    start := time.Now()
    result := PipelineResult{ProcessorName: pipeline.Name()}

    logger.InfoContext(ctx, "pipeline starting", "name", pipeline.Name())

    // Step 1: Validate (fixed — always present)
    if err := pipeline.Validate(ctx); err != nil {
        result.Error = fmt.Errorf("validation failed: %w", err)
        pipeline.OnError(ctx, result.Error)
        return result
    }

    // Step 2: Extract (fixed — always present)
    records, err := pipeline.Extract(ctx)
    if err != nil {
        result.Error = fmt.Errorf("extract failed: %w", err)
        pipeline.OnError(ctx, result.Error)
        return result
    }
    result.RecordsRead = len(records)
    logger.InfoContext(ctx, "extract complete", "records", len(records))

    // Step 3: Transform (variable — default implementation = pass-through)
    transformed, err := pipeline.Transform(ctx, records)
    if err != nil {
        result.Error = fmt.Errorf("transform failed: %w", err)
        pipeline.OnError(ctx, result.Error)
        return result
    }
    logger.InfoContext(ctx, "transform complete", "records", len(transformed))

    // Step 4: Load (fixed — always present)
    written, err := pipeline.Load(ctx, transformed)
    if err != nil {
        result.Error = fmt.Errorf("load failed: %w", err)
        pipeline.OnError(ctx, result.Error)
        return result
    }
    result.RecordsWritten = written
    result.Duration = time.Since(start)

    logger.InfoContext(ctx, "pipeline complete",
        "name", pipeline.Name(),
        "read", result.RecordsRead,
        "written", result.RecordsWritten,
        "duration_ms", result.Duration.Milliseconds(),
    )

    pipeline.OnSuccess(ctx, result)
    return result
}

package pipeline

import (
    "context"
    "database/sql"
    "encoding/csv"
    "fmt"
    "os"
    "strings"
    "time"
)

// CSVToDBPipeline reads a CSV and stores it into the database.
// It implements Transform to clean the data.
type CSVToDBPipeline struct {
    BasePipeline                     // embed to get the default hooks
    FilePath   string
    TableName  string
    DB         *sql.DB
    NotifySvc  NotificationService   // for the OnSuccess hook
}

func (p *CSVToDBPipeline) Name() string { return "csv-to-db:" + p.FilePath }

func (p *CSVToDBPipeline) Validate(ctx context.Context) error {
    if _, err := os.Stat(p.FilePath); os.IsNotExist(err) {
        return fmt.Errorf("file not found: %s", p.FilePath)
    }
    if p.TableName == "" {
        return fmt.Errorf("table name is required")
    }
    if err := p.DB.PingContext(ctx); err != nil {
        return fmt.Errorf("database connection failed: %w", err)
    }
    return nil
}

func (p *CSVToDBPipeline) Extract(ctx context.Context) ([]Record, error) {
    file, err := os.Open(p.FilePath)
    if err != nil {
        return nil, fmt.Errorf("cannot open file: %w", err)
    }
    defer file.Close()

    reader := csv.NewReader(file)
    rows, err := reader.ReadAll()
    if err != nil {
        return nil, fmt.Errorf("cannot read CSV: %w", err)
    }

    if len(rows) < 2 {
        return nil, fmt.Errorf("CSV must have header row and at least one data row")
    }

    headers := rows[0]
    records := make([]Record, 0, len(rows)-1)
    for _, row := range rows[1:] {
        record := make(Record)
        for i, value := range row {
            if i < len(headers) {
                record[headers[i]] = value
            }
        }
        records = append(records, record)
    }

    return records, nil
}

// Transform cleans the data: trim whitespace, normalize empty fields.
// Overrides the BasePipeline hook method — this is what differs from other pipelines.
func (p *CSVToDBPipeline) Transform(ctx context.Context, records []Record) ([]Record, error) {
    cleaned := make([]Record, 0, len(records))
    for _, record := range records {
        clean := make(Record)
        skip := false
        for k, v := range record {
            str, ok := v.(string)
            if !ok {
                clean[k] = v
                continue
            }
            str = strings.TrimSpace(str)
            // Skip the record if a key field is empty
            if k == "id" && str == "" {
                skip = true
                break
            }
            clean[k] = str
        }
        if !skip {
            clean["imported_at"] = time.Now().Format(time.RFC3339)
            cleaned = append(cleaned, clean)
        }
    }
    return cleaned, nil
}

func (p *CSVToDBPipeline) Load(ctx context.Context, records []Record) (int, error) {
    if len(records) == 0 {
        return 0, nil
    }

    tx, err := p.DB.BeginTx(ctx, nil)
    if err != nil {
        return 0, fmt.Errorf("cannot begin transaction: %w", err)
    }
    defer tx.Rollback()

    count := 0
    for _, record := range records {
        // Build the INSERT query from the record keys
        columns := make([]string, 0, len(record))
        placeholders := make([]string, 0, len(record))
        values := make([]interface{}, 0, len(record))
        i := 1
        for col, val := range record {
            columns = append(columns, col)
            placeholders = append(placeholders, fmt.Sprintf("$%d", i))
            values = append(values, val)
            i++
        }

        query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s) ON CONFLICT DO NOTHING",
            p.TableName,
            strings.Join(columns, ", "),
            strings.Join(placeholders, ", "),
        )

        if _, err := tx.ExecContext(ctx, query, values...); err != nil {
            return count, fmt.Errorf("insert failed: %w", err)
        }
        count++
    }

    if err := tx.Commit(); err != nil {
        return 0, fmt.Errorf("commit failed: %w", err)
    }

    return count, nil
}

// OnSuccess override: send a notification after success.
func (p *CSVToDBPipeline) OnSuccess(ctx context.Context, result PipelineResult) {
    if p.NotifySvc == nil {
        return
    }
    msg := fmt.Sprintf("Pipeline %s finished: %d records imported into %s in %v",
        result.ProcessorName, result.RecordsWritten, p.TableName, result.Duration)
    _ = p.NotifySvc.Notify(ctx, "[email protected]", msg)
}

// APIToStoragePipeline fetches data from a REST API and stores it in cloud storage.
// It does not implement Transform (uses the default pass-through from BasePipeline).
type APIToStoragePipeline struct {
    BasePipeline                         // no Transform needed — data is stored as-is
    APIEndpoint  string
    APIKey       string
    BucketName   string
    OutputPath   string
    HTTPClient   *http.Client
    Storage      CloudStorage
}

func (p *APIToStoragePipeline) Name() string {
    return fmt.Sprintf("api-to-storage:%s", p.APIEndpoint)
}

func (p *APIToStoragePipeline) Validate(ctx context.Context) error {
    if p.APIEndpoint == "" {
        return fmt.Errorf("API endpoint is required")
    }
    if p.BucketName == "" {
        return fmt.Errorf("bucket name is required")
    }
    // Test the connection to the API
    req, err := http.NewRequestWithContext(ctx, http.MethodHead, p.APIEndpoint, nil)
    if err != nil {
        return fmt.Errorf("invalid API endpoint: %w", err)
    }
    req.Header.Set("Authorization", "Bearer "+p.APIKey)
    resp, err := p.HTTPClient.Do(req)
    if err != nil {
        return fmt.Errorf("API unreachable: %w", err)
    }
    defer resp.Body.Close()
    if resp.StatusCode == http.StatusUnauthorized {
        return fmt.Errorf("invalid API key")
    }
    return nil
}

func (p *APIToStoragePipeline) Extract(ctx context.Context) ([]Record, error) {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, p.APIEndpoint, nil)
    if err != nil {
        return nil, fmt.Errorf("cannot build request: %w", err)
    }
    req.Header.Set("Authorization", "Bearer "+p.APIKey)

    resp, err := p.HTTPClient.Do(req)
    if err != nil {
        return nil, fmt.Errorf("API call failed: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return nil, fmt.Errorf("API returned status %d", resp.StatusCode)
    }

    var records []Record
    if err := json.NewDecoder(resp.Body).Decode(&records); err != nil {
        return nil, fmt.Errorf("cannot decode response: %w", err)
    }
    return records, nil
}

// Transform is NOT implemented — it uses the default from BasePipeline (pass-through).
// Data from the API is stored directly without transformation.

func (p *APIToStoragePipeline) Load(ctx context.Context, records []Record) (int, error) {
    data, err := json.Marshal(records)
    if err != nil {
        return 0, fmt.Errorf("cannot marshal records: %w", err)
    }

    key := fmt.Sprintf("%s/%s.json", p.OutputPath, time.Now().Format("2006-01-02T15-04-05"))
    if err := p.Storage.Upload(ctx, p.BucketName, key, data, "application/json"); err != nil {
        return 0, fmt.Errorf("upload failed: %w", err)
    }

    return len(records), nil
}

func main() {
    ctx := context.Background()
    logger := slog.Default()

    // Two different pipelines, but Execute is called the same way
    pipelines := []DataPipeline{
        &CSVToDBPipeline{
            FilePath:  "data/users.csv",
            TableName: "users",
            DB:        db,
            NotifySvc: notifSvc,
        },
        &APIToStoragePipeline{
            APIEndpoint: "https://api.external.com/v1/products",
            APIKey:      "sk_live_xxxx",
            BucketName:  "raw-data",
            OutputPath:  "products",
            HTTPClient:  &http.Client{Timeout: 30 * time.Second},
            Storage:     gcsStorage,
        },
    }

    for _, p := range pipelines {
        result := Execute(ctx, p, logger)
        if result.Error != nil {
            log.Printf("Pipeline %s failed: %v", result.ProcessorName, result.Error)
            continue
        }
        log.Printf("Pipeline %s: read=%d written=%d duration=%v",
            result.ProcessorName, result.RecordsRead, result.RecordsWritten, result.Duration)
    }
}

package report

import "context"

// ReportData stores the data to be formatted into a report.
type ReportData struct {
    Title     string
    Period    string
    Rows      []map[string]interface{}
    Summary   map[string]interface{}
}

// ReportGenerator defines the steps that vary per format.
type ReportGenerator interface {
    Name() string
    FetchData(ctx context.Context, period string) (*ReportData, error)
    Render(data *ReportData) ([]byte, error)
    ContentType() string
    FileExtension() string
    // Hook: before render (optional — default no-op)
    PreRender(data *ReportData) *ReportData
}

// BaseReportGenerator provides the default hook.
type BaseReportGenerator struct{}

func (b *BaseReportGenerator) PreRender(data *ReportData) *ReportData {
    return data // default: no modification
}

// GenerateReport is the Template Method for report generation.
func GenerateReport(ctx context.Context, gen ReportGenerator, period, outputDir string) (string, error) {
    // Step 1: Fetch the data (variable)
    data, err := gen.FetchData(ctx, period)
    if err != nil {
        return "", fmt.Errorf("fetch data failed: %w", err)
    }

    // Step 2: Pre-render hook (variable/optional)
    data = gen.PreRender(data)

    // Step 3: Validate the data (fixed — the same for all)
    if len(data.Rows) == 0 {
        return "", fmt.Errorf("no data available for period: %s", period)
    }

    // Step 4: Render (variable — PDF, Excel, HTML differ)
    content, err := gen.Render(data)
    if err != nil {
        return "", fmt.Errorf("render failed: %w", err)
    }

    // Step 5: Save to a file (fixed — the same for all)
    filename := fmt.Sprintf("%s/%s_%s.%s",
        outputDir,
        strings.ToLower(strings.ReplaceAll(data.Title, " ", "_")),
        period,
        gen.FileExtension(),
    )
    if err := os.WriteFile(filename, content, 0644); err != nil {
        return "", fmt.Errorf("save failed: %w", err)
    }

    return filename, nil
}


// PDFReportGenerator implements rendering to PDF.
type PDFReportGenerator struct {
    BaseReportGenerator
    db      *sql.DB
    pdfSvc  PDFService
}

func (g *PDFReportGenerator) Name() string         { return "PDF Report Generator" }
func (g *PDFReportGenerator) ContentType() string  { return "application/pdf" }
func (g *PDFReportGenerator) FileExtension() string { return "pdf" }

func (g *PDFReportGenerator) FetchData(ctx context.Context, period string) (*ReportData, error) {
    // Query specific to the monthly PDF report
    rows, err := g.db.QueryContext(ctx,
        "SELECT date, revenue, expenses FROM financials WHERE period = $1 ORDER BY date", period)
    if err != nil {
        return nil, err
    }
    defer rows.Close()
    // ... scan rows ...
    return &ReportData{Title: "Financial Report", Period: period}, nil
}

func (g *PDFReportGenerator) Render(data *ReportData) ([]byte, error) {
    return g.pdfSvc.GeneratePDF(data.Title, data.Rows, data.Summary)
}


// HTMLReportGenerator implements rendering to HTML.
type HTMLReportGenerator struct {
    BaseReportGenerator
    db          *sql.DB
    tmplPath    string
}

func (g *HTMLReportGenerator) Name() string         { return "HTML Report Generator" }
func (g *HTMLReportGenerator) ContentType() string  { return "text/html" }
func (g *HTMLReportGenerator) FileExtension() string { return "html" }

func (g *HTMLReportGenerator) FetchData(ctx context.Context, period string) (*ReportData, error) {
    // A different query for the HTML report (possibly more detailed)
    return &ReportData{Title: "Financial Report", Period: period}, nil
}

func (g *HTMLReportGenerator) Render(data *ReportData) ([]byte, error) {
    tmpl, err := template.ParseFiles(g.tmplPath)
    if err != nil {
        return nil, fmt.Errorf("cannot parse template: %w", err)
    }
    var buf bytes.Buffer
    if err := tmpl.Execute(&buf, data); err != nil {
        return nil, fmt.Errorf("render failed: %w", err)
    }
    return buf.Bytes(), nil
}

// Override the PreRender hook — add computed fields before rendering
func (g *HTMLReportGenerator) PreRender(data *ReportData) *ReportData {
    // Calculate the total for the summary
    var total float64
    for _, row := range data.Rows {
        if rev, ok := row["revenue"].(float64); ok {
            total += rev
        }
    }
    data.Summary["grand_total"] = total
    return data
}

// Template Method: the step order is fixed, some steps differ
// One algorithm, variation inside it
func Execute(p DataPipeline) error {
    p.Validate()     // always first
    p.Extract()      // always second
    p.Transform()    // always third (may have a different implementation)
    p.Load()         // always fourth
    // The order CANNOT be changed by implementations
}

// Strategy: the whole algorithm can differ
// Many algorithms, the client picks which one
type ShippingStrategy interface {
    Calculate(order Order) (int, error)
}
// JNE, J&T, GoSend — completely different calculation algorithms
// The client chooses the strategy; no step order is forced

// DataPipeline with a Transform that uses the Strategy Pattern
type FlexiblePipeline struct {
    BasePipeline
    source      DataSource        // strategy for Extract
    transformer DataTransformer   // strategy for Transform
    sink        DataSink          // strategy for Load
}

// Run is the Template Method — fixed order
func (p *FlexiblePipeline) Run(ctx context.Context) error {
    // Fixed step: log start
    log.Printf("Pipeline starting")

    // Variable step (via strategy): fetch the data
    data, err := p.source.Read(ctx)
    if err != nil {
        return err
    }

    // Variable step (via strategy): transform
    transformed, err := p.transformer.Transform(ctx, data)
    if err != nil {
        return err
    }

    // Variable step (via strategy): save
    if err := p.sink.Write(ctx, transformed); err != nil {
        return err
    }

    // Fixed step: log completion
    log.Printf("Pipeline complete")
    return nil
}

// Usage: swap strategies without changing the order
pipeline := &FlexiblePipeline{
    source:      &CSVSource{filePath: "data.csv"},
    transformer: &CleaningTransformer{removeNulls: true},
    sink:        &PostgreSink{db: db, table: "users"},
}
pipeline.Run(ctx)

// Swap the sink to cloud storage — the order stays the same
pipeline.sink = &S3Sink{bucket: "raw-data", key: "users.json"}
pipeline.Run(ctx)

// MockPipeline for testing the Execute template method
type MockPipeline struct {
    BasePipeline
    name          string
    validateErr   error
    extractData   []Record
    extractErr    error
    transformData []Record
    transformErr  error
    loadCount     int
    loadErr       error
    successCalled bool
    errorCalled   bool
}

func (m *MockPipeline) Name() string { return m.name }

func (m *MockPipeline) Validate(ctx context.Context) error {
    return m.validateErr
}

func (m *MockPipeline) Extract(ctx context.Context) ([]Record, error) {
    return m.extractData, m.extractErr
}

func (m *MockPipeline) Transform(ctx context.Context, records []Record) ([]Record, error) {
    if m.transformData != nil {
        return m.transformData, m.transformErr
    }
    return records, m.transformErr
}

func (m *MockPipeline) Load(ctx context.Context, records []Record) (int, error) {
    if m.loadErr != nil {
        return 0, m.loadErr
    }
    m.loadCount = len(records)
    return m.loadCount, nil
}

func (m *MockPipeline) OnSuccess(ctx context.Context, result PipelineResult) {
    m.successCalled = true
}

func (m *MockPipeline) OnError(ctx context.Context, err error) {
    m.errorCalled = true
}


func TestExecute_HappyPath(t *testing.T) {
    mock := &MockPipeline{
        name:        "test-pipeline",
        extractData: []Record{{"id": "1", "name": "Alice"}, {"id": "2", "name": "Bob"}},
    }

    result := Execute(context.Background(), mock, slog.Default())

    if result.Error != nil {
        t.Fatalf("expected no error, got: %v", result.Error)
    }
    if result.RecordsRead != 2 {
        t.Errorf("expected 2 records read, got %d", result.RecordsRead)
    }
    if result.RecordsWritten != 2 {
        t.Errorf("expected 2 records written, got %d", result.RecordsWritten)
    }
    if !mock.successCalled {
        t.Error("OnSuccess should have been called")
    }
    if mock.errorCalled {
        t.Error("OnError should NOT have been called")
    }
}

func TestExecute_StopsAtValidationError(t *testing.T) {
    mock := &MockPipeline{
        name:        "test-pipeline",
        validateErr: fmt.Errorf("config missing"),
    }

    result := Execute(context.Background(), mock, slog.Default())

    if result.Error == nil {
        t.Error("expected error from validation failure")
    }
    if result.RecordsRead != 0 {
        t.Error("no records should be read after validation failure")
    }
    if !mock.errorCalled {
        t.Error("OnError should have been called on validation failure")
    }
}

func TestExecute_StopsAtExtractError(t *testing.T) {
    mock := &MockPipeline{
        name:       "test-pipeline",
        extractErr: fmt.Errorf("source unavailable"),
    }

    result := Execute(context.Background(), mock, slog.Default())

    if result.Error == nil {
        t.Error("expected error from extract failure")
    }
    if mock.loadCount != 0 {
        t.Error("Load should not be called after Extract failure")
    }
}

func TestCSVToDBPipeline_Transform_SkipsEmptyID(t *testing.T) {
    pipeline := &CSVToDBPipeline{}
    records := []Record{
        {"id": "1", "name": "Alice "},
        {"id": "", "name": "Bob"},  // must be skipped because the ID is empty
        {"id": "3", "name": " Charlie"},
    }

    result, err := pipeline.Transform(context.Background(), records)
    if err != nil {
        t.Fatalf("unexpected error: %v", err)
    }
    if len(result) != 2 {
        t.Errorf("expected 2 records after skipping empty ID, got %d", len(result))
    }
    // Verify the trimming
    if result[0]["name"] != "Alice" {
        t.Errorf("expected trimmed name 'Alice', got %q", result[0]["name"])
    }
}