Form

Send lists on from submit

    list := r.Form["something[]"]
    // Aggregate all form inputs with the same name attribute

Testing

Run tests a designated number of times, ignoring no file changes optimization

go test -count=1 ./...

Examples

Performance Profiling Setup with pprof

This document explains how performance profiling was integrated into the BOMR Forecaster application using Go's runtime/pprof package.

Overview

The application now includes HTTP endpoints for runtime profiling, allowing you to identify CPU and memory hot paths in your codebase. This is enabled via the net/http/pprof package, which exposes profiling endpoints over HTTP.

Changes Made

1. Router Changes (router/router.go)

Import Added:

import (
    // ... existing imports
    "net/http/pprof"
)

Handler Registration: Added pprof endpoint registration in the New() function (lines 137-151):

// Enable pprof endpoints in local mode, dev environment, or when explicitly enabled
if config.EnablePprof {
    // Register pprof handlers manually since we're using a custom ServeMux
    r.mux.HandleFunc("/debug/pprof/", pprof.Index)
    r.mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
    r.mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
    r.mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
    r.mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
    r.mux.HandleFunc("/debug/pprof/heap", pprof.Handler("heap").ServeHTTP)
    r.mux.HandleFunc("/debug/pprof/goroutine", pprof.Handler("goroutine").ServeHTTP)
    r.mux.HandleFunc("/debug/pprof/allocs", pprof.Handler("allocs").ServeHTTP)
    r.mux.HandleFunc("/debug/pprof/block", pprof.Handler("block").ServeHTTP)
    r.mux.HandleFunc("/debug/pprof/mutex", pprof.Handler("mutex").ServeHTTP)
    log.Printf("pprof endpoints enabled at /debug/pprof/")
}

Key Points:

2. Config Changes (config/config.go)

New Field Added:

type Config struct {
    // ... existing fields
    EnablePprof bool `json:"ENABLE_PPROF"`
    // ... rest of fields
}

Environment Variable Support: Added check in New() function (lines 112-115):

// Check environment variable as override for pprof
if os.Getenv("ENABLE_PPROF") == "true" {
    C.EnablePprof = true
}

Key Points:

3. Makefile Changes (Makefile)

New Target Added: // note when starting up this server you may see this message:

Couldn't find a suitable web browser! Set the BROWSER environment variable to your desired browser.

This likely means that you are running on headless machine, which is fine because the only thing that failed was the browser auto-starting. You can still connect to the server.

p:
	@echo "Starting application with pprof enabled..."
	@echo "pprof endpoints will be available at: http://localhost:8081/debug/pprof/"
	@echo "Example: go tool pprof http://localhost:8081/debug/pprof/profile?seconds=30"
	$(eval include include .env.local)
	ENABLE_PPROF=true go run .

Key Points:

Usage

Starting the Application with Profiling

make p

This will:

  1. Start the application on the configured port (default: 8081)
  2. Enable pprof endpoints at /debug/pprof/
  3. Log a message confirming pprof is enabled

Available Endpoints

Once running, the following endpoints are available:

Generating Profiles

CPU Profile (30 seconds)
go tool pprof http://localhost:8081/debug/pprof/profile?seconds=30

Important: Generate load on your application while profiling is running, otherwise you'll get 0 samples.

Heap/Memory Profile
go tool pprof http://localhost:8081/debug/pprof/heap
Goroutine Profile
go tool pprof http://localhost:8081/debug/pprof/goroutine

Analyzing Profiles

Interactive Mode

Once in pprof interactive mode, use these commands:

Web UI (Recommended)

For the easiest analysis experience:

# Save profile to file first
curl "http://localhost:8081/debug/pprof/profile?seconds=30" > cpu.prof

# Open in web UI
go tool pprof -http=:8082 cpu.prof

# Store profile data in memory so you don't have to remember to clean up
go tool pprof -http=:8082 "http://localhost:8081/debug/pprof/profile?seconds=30"

This opens a web interface at http://localhost:8082 with:

Example Workflow

  1. Start application with profiling:

    make p

  2. In another terminal, generate load while profiling:

    # Generate a 30-second CPU profile
go tool pprof http://localhost:8081/debug/pprof/profile?seconds=30

# While profiling is running, use your application normally
# or generate load with curl requests

  3. Analyze the profile:

    # If you saved it to a file
go tool pprof -http=:8082 cpu.prof

# Or use interactive mode
go tool pprof cpu.prof
(pprof) top10
(pprof) list <function_name>
(pprof) web

Prerequisites

Install Graphviz (for visual graphs)

Ubuntu/Debian:

sudo apt-get update
sudo apt-get install graphviz

macOS:

brew install graphviz

Fedora/RHEL:

sudo dnf install graphviz

Verify installation:

dot -V

Troubleshooting

"No samples were found"

This means your application was idle during profiling. Solutions:

  1. Generate load while profiling:

    # Terminal 1: Start profiling
go tool pprof http://localhost:8081/debug/pprof/profile?seconds=30

# Terminal 2: Generate load
for i in {1..100}; do
    curl -s http://localhost:8081/bomr/getForecast > /dev/null &
done

  2. Use different profile types:

    • Heap profile works even when idle
    • Goroutine profile shows what goroutines are doing

  3. Enable block profiling (for I/O-bound apps): Add to main.go:

    import _ "runtime"

func main() {
    runtime.SetBlockProfileRate(1)
    runtime.SetMutexProfileFraction(1)
    // ... rest of code
}

Profile file is binary

CPU profile files (.pb.gz) are binary format. Use go tool pprof to read them:

go tool pprof cpu.prof
# or
go tool pprof -http=:8082 cpu.prof

Port conflicts

If port 8081 is already in use, check your config file for the HTTP_PORT setting and adjust accordingly.

Security Considerations

Important: Never expose pprof endpoints in production without proper authentication, as they can reveal sensitive information about your application's internals.

Additional Resources

Summary

The pprof integration provides:

Use make p to start profiling, then use go tool pprof to analyze your application's performance characteristics and identify hot paths.

Async

var wg sync.WaitGroup
errChan := make(chan error)
var results sync.Map

//	loop example:
for key, value := range someMap {
    wg.Add(1) // Increment the wait group counter
    go func(k, v string) {
        defer wg.Done() // Decrement the wait group counter when done
        item, e := someMethod(k, v)
        if e != nil {
            errChan <- fmt.Errorf("error when someMethod for someParentMethod() of key: %s. Error: %v", k, e)
            return
        }
        results.Store(k, item)
    }(key, value)
}

// inline example:
wg.Add(1)
go func() {
    defer wg.Done()
    var e error
    someResult, e = someMethod()
    if e != nil {
        errChan <- fmt.Errorf("error, when someMethod() for someParentMethod(). Error: %v", e)
        return
    }
    results.Store(someMapKey, someResult)
}()

go func() {
    wg.Wait()
    close(errChan)
}()

if errChanError := <-errChan; errChanError != nil {
    return fmt.Errorf("error, when attempting to perform async actions. Error: %v", errChanError)
}

Cancel long running job

func DoingSOmething() error {
    done := make(chan error, 1)

	go func() {
            err := doingSomethingElse()
            if err != nil {
                done <- fmt.Errorf("error, when doingSomethingElse() for DoingSomething(). Error: %v", err)
                return
            }
	    done <- nil
	}()

	select {
	case err := <-done:
		return err
	case <- time.After(20 * time.Seconds):
		return errors.New("error, doingSOmethingElse() for DoingSomething() has timed out")
	}
}

DB Backup

func StartDailyBackup(ctx context.Context, dbFileLocation string) error {
	location, err := time.LoadLocation("America/Chicago") // familiar timezone for an easier schedule
	if err != nil {
		return fmt.Errorf("error, invalid location provided. Error: %v", err)
	}
	ticker := time.NewTicker(time.Hour)

	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			currentTime := time.Now()
			localTime := currentTime.In(location)
			h, _, _ := localTime.Clock()
			if h == 20 { // 8 pm 
				err := backupDatabase(ctx, dbFileLocation)
				if err != nil {
					return fmt.Errorf("error, when backupDatabase() for StartDailyBackup(). Error: %v", err)
				}
			}
		}
	}
}


func backupDatabase(ctx context.Context) error {
	log.Printf("starting database backup")
	backupFile := "/data/database_backup"
    statement := fmt.Sprintf("VACUUM main INTO '%s'", backupFile)
	_, err := database.Exec(statement)
	if err != nil {
        return fmt.Errorf("error, when backing up the database. Error: %v", err)
	}
	fileBytes, err := os.ReadFile(backupFile)
	if err != nil {
		return fmt.Errorf("error, when reading database backup file. Error: %v", err)
	}
	uReq := UploadToBucketRequest{
		bucketName: EnvironmentStateBucketName,
		fileName:   "database_backup",
		data:       fileBytes,
	}
	err = uploadToBucket(
		ctx,
		uReq,
	)
	if err != nil {
		return fmt.Errorf("error, when uploadToBucket() for backupDatabase(). Error: %v", err)
	}
	log.Printf("database backup complete")
	return nil
}


type UploadToBucketRequest struct {
	bucketName string
	fileName   string
	data       []byte
}

// uploadToBucket assuming GCP for example
func uploadToBucket(
	ctx context.Context,
	req UploadToBucketRequest,
) error {
	client, err := storage.NewClient(ctx)
	if err != nil {
		return fmt.Errorf("error, failed to create GCP storage client. Error: %v", err)
	}
	bucket := client.Bucket(req.bucketName)
	object := bucket.Object(req.fileName)
	customTime := time.Now()
	writer := object.NewWriter(ctx)
	writer.CustomTime = customTime
	reader := bytes.NewReader(req.data)
	if _, err = io.Copy(writer, reader); err != nil {
		return fmt.Errorf("error, failed to upload %s to GCP bucket: %s. Error: %v", req.fileName, req.bucketName, err)
	}
	if err = writer.Close(); err != nil {
		return fmt.Errorf("error, failed to close file writer. Error: %v", err)
	}
	return nil
}

// downloadFromBucket assuming GCP for example
func downloadFromBucket(
	ctx context.Context,
	bucketName,
	objectName string,
	generation int64,
) (bucketStuff []byte, err error) {
	var closeErr error
	// todo look into reusing the storage client
	client, err := storage.NewClient(ctx)
	if err != nil {
		return nil, fmt.Errorf("error, when attempting to create new GCP storage client. Error: %v", err)
	}
	defer func(client *storage.Client) {
		closeErr = client.Close()
		if closeErr != nil {
			err = fmt.Errorf("error, when attempting to close the GCP storage client. Error: %w", closeErr)
			return
		}
	}(client)

	bucket := client.Bucket(bucketName)
	var object *storage.ObjectHandle
	if generation == 0 {
		object = bucket.Object(objectName)
	} else {
		object = bucket.Object(objectName).Generation(generation)
	}

	rc, err := object.NewReader(ctx)
	if err != nil {
		return nil, fmt.Errorf("error, when attempting to create new reader from gcp bucket object. Error: %w", err)
	}

	defer func(rc *storage.Reader) {
		closeErr = rc.Close()
		if closeErr != nil {
			err = fmt.Errorf("error, when attempting to close object reader. Error: %w", closeErr)
			return
		}
	}(rc)

	data, err := io.ReadAll(rc)
	if err != nil {
		return nil, fmt.Errorf("error, when attempting to read from object reader. Error: %v", err)
	}
	return data, nil
}

db log helper

// log exact query used for debugging queries
func FormatQuery(query string, args ...interface{}) string {
	for _, arg := range args {
		var formatted string
		switch v := arg.(type) {
		case string:
			formatted = fmt.Sprintf("'%v'", v)
		case sql.NullString:
			if v.Valid {
				formatted = fmt.Sprintf("'%v'", v.String)
			} else {
				formatted = "NULL"
			}
		default:
			formatted = fmt.Sprintf("%v", v)
		}
		query = strings.Replace(query, "?", formatted, 1)
	}
	return base64.StdEncoding.EncodeToString([]byte(query))
}

// Example:
if err != nil {
    theStatement = shared.FormatQuery(theStatement, args...)
    return nil, fmt.Errorf("error, when attempting to retrieve records. Query: %s. Error: %v", theStatement, err)
}

db transaction

ctx := context.TODO()
tx, err := _CONNECTION_POOL_.Begin(ctx)
if err != nil {
    return fmt.Errorf("error, when attempting to start a transaction. Error: %v", err)
}

err = func() error {
    _, err2 := tx.Exec(ctx, `_QUERY_HERE_`)
    if err2 != nil {
        return fmt.Errorf("error, when executing query to _do_something_. Error: %v", err2)
    }

    _, err2 = tx.Exec(ctx,  `_QUERY_HERE_`)
    if err2 != nil {
        return fmt.Errorf("error, when executing query to _do_something_. Error: %v", err2)
    }

    _, err2 = tx.Exec(ctx, `_QUERY_HERE_`)
    if err2 != nil {
        return fmt.Errorf("error, when executing query to _do_something_. Error: %v", err2)
    }
    return nil
}()
if err != nil {
    rollBackErr := tx.Rollback(ctx)
    if rollBackErr != nil {
        return fmt.Errorf("error, when attempting to roll back commit: Rollback Error: %v, Original Error: %v", rollBackErr, err)
    }
    return fmt.Errorf("error, when attempting to perform database transaction. Error: %v", err)
}
err = tx.Commit(ctx)
if err != nil {
    return fmt.Errorf("error, when attempting to commit the transaction to the database. Error: %v", err)
}

Fetch Row

var result string
err = _DATABASE_CONNECTION_POOL_.QueryRow(
    `SELECT some_column
    FROM some_table
    WHERE some_column = ?
      AND some_other_column IS NULL
      ORDER BY id DESC
      LIMIT 1`,
    someArg,
).Scan(
    &result,
)
if err != nil {
    if errors.Is(err, sql.ErrNoRows) {
        // todo implement
    } else {
        return nil, fmt.Errorf("error, when attempting to execute sql statement. Error: %w", err)
    }
}

Fetch Rows

rows, err := _DATABASE_CONNECTION_POOL_.Query(
    `SELECT st.id, st.some_column
    FROM some_table st
    JOIN some_other_table sot ON st.id = sot.id
    WHERE sot.some_column = ?`,
    someArg,
)

defer rows.Close()

if err != nil {
    return nil, fmt.Errorf("error, when attempting to retrieve records. Error: %v", err)
}

var result []_RESULT_STRUCT_
for rows.Next() {
    var r _RESULT_STRUCT_
    err = rows.Scan(
        &r.Id,
        &r._SOME_PROPERTY_,
    )
    if err != nil {
        return nil, fmt.Errorf("error, when scanning database rows. Error: %v", err)
    }
    result = append(result, r)
}

err = rows.Err()
if err != nil {
    return nil, fmt.Errorf("error, when iterating through database rows. Error: %v", err)
}
return result, nil

Smart redirect

func SmartRedirect(w http.ResponseWriter, r *http.Request, u string) {
	if r.Header.Get("HX-Request") == "true" { // was triggered from button press
		w.Header().Set("HX-Redirect", u)
	} else { //  was triggered from page refresh
		http.Redirect(w, r, u, http.StatusSeeOther)
	}
}

HTTP Request

package template_golang

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
)


// http.MethodPost
// http.MethodGet
// http.MethodPut
// http.MethodDelete
func HttpRequest(method, theURL string, reqBody any, dst any) error {
	var body io.Reader
	if reqBody != nil {
		buf := new(bytes.Buffer)
		if err := json.NewEncoder(buf).Encode(reqBody); err != nil {
			return fmt.Errorf("error, %s %q: encoding request body: %w", method, theURL, err)
		}
		body = buf
	}

	request, err := http.NewRequest(method, theURL, body)
	if err != nil {
		return fmt.Errorf("error, when generating get request: %w", err)
	}
	request.Header.Set("Content-Type", "application/json")

	resp, err := http.DefaultClient.Do(request)
	if err != nil {
		return fmt.Errorf("error, request if http method %s failed: %w", method, err)
	}
	defer resp.Body.Close()

	if resp.StatusCode < 200 || resp.StatusCode > 299 {
		const maxErrorBody = 4 << 10 // 4 KiB
		rb, readErr := io.ReadAll(&io.LimitedReader{R: resp.Body, N: maxErrorBody})
		if readErr != nil {
			return fmt.Errorf("error, when reading error resp body: %w", readErr)
		}
		if resp.StatusCode == http.StatusNotFound {
			log.Printf("received a 404 when attempting url. Url: %s", request.URL)
		}
		return fmt.Errorf(
			"error, when performing get request. REQUEST METHOD: %s. RESPONSE CODE: %d. RESPONSE MESSAGE: %s",
			method,
			resp.StatusCode,
			string(rb),
		)
	}

	// Caller doesn't care about response body; just drain so connection can be reused.
	if dst == nil {
		_, _ = io.Copy(io.Discard, resp.Body)
		return nil
	}

	dec := json.NewDecoder(resp.Body)
	if err := dec.Decode(dst); err != nil {
		if err == io.EOF {
			return fmt.Errorf("%s %q: empty response body", method, theURL)
		}
		return fmt.Errorf("error, %s %q: decoding JSON response: %w", method, theURL, err)
	}

	return nil
}

Insert Record

_, err := _DATABASE_CONNECTION_POOL_.Exec(
    "INSERT INTO _SOME_TABLE_ (_SOME_COLUMN_, _SOME_COLUMN_) VALUES ($1, $2)",
    r._SOME_PROPERTY_,
    r._SOME_PROPERTY_,
)
if err != nil {
    return fmt.Errorf("error, when attempting to persist a request for a stress test: %v", err)
}

Lock Insertion

// In migration file:
// -- Set busy_timeout to 5000 milliseconds (5 seconds)
// -- This ensures SQLite will wait up to 5 seconds for a lock before returning SQLITE_BUSY
// PRAGMA busy_timeout = 5000;
// PRAGMA journal_mode = WAL;

// -- Create a test table for health check lock testing
// -- This table is used to test database lock detection
// CREATE TABLE IF NOT EXISTS health_check_test (
//     id INTEGER PRIMARY KEY,
//     test_value TEXT
// );



// CheckDatabaseLock checks if the database is locked for more than 5 seconds.
// With PRAGMA busy_timeout = 5000, SQLite will only return SQLITE_BUSY after waiting 5 seconds.
// Returns an error if SQLITE_BUSY is encountered, indicating the database has been locked for more than 5 seconds.
func (m *HealthModel) CheckDatabaseLock(ctx context.Context, timeout time.Duration) error {
	// Create a context with timeout for the database check
	checkCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	// Attempt to acquire a write lock immediately
	// With PRAGMA busy_timeout = 5000, this will wait up to 5 seconds before returning SQLITE_BUSY
	_, err := m.database..ExecContext(checkCtx, "DELETE FROM health_check_test WHERE id = 999")
	if err != nil {
		// Check if the error is a SQLite error with BUSY_TIMEOUT or BUSY code
		var sqliteErr *sqlite3.Error
		if errors.As(err, &sqliteErr) {
			// Check for BUSY_TIMEOUT (extended error code) or BUSY (base error code)
			if sqliteErr.ExtendedCode() == sqlite3.BUSY_TIMEOUT || sqliteErr.Code() == sqlite3.BUSY {
				msg := fmt.Sprintf("database has been locked for more than 5 seconds: %v", err)
				log.Println(msg)
				return errors.New(msg)
			}
		}
		// Other errors (connection issues, etc.)
		msg := fmt.Sprintf("database error: %v", err)
		log.Println(msg)
		return errors.New(msg)
	}

	// Successfully acquired the lock within 5 seconds
	return nil
}

Mutex Profiler

type ProfilerMutex struct {
	mu                               sync.Mutex
	startTime time.Time
	queryName string
}

func (pm *ProfilerMutex) Lock(queryName string) {
	pm.mu.Lock()
	pm.startTime = time.Now()
	pm.queryName = queryName
}

func (pm *ProfilerMutex) Unlock() {
	log.Printf("%s. Duration: %s\n", pm.queryName, time.Since(pm.startTime))
	pm.mu.Unlock()
}

nats client

import (
	"github.com/nats-io/nats.go"
)


opts := []nats.Option{
    nats.MaxReconnects(-1),
	nats.Token(config.DbExpressAgentKey),
	nats.ProxyPath("/nats/wss"), // only applicable to websockets (unconfirmed), but this configuration has to be on the client side for some reason. Cannot put the path in the URL, the nats url can only take host and port, no paths allowed
}
// url can be any of these as long as the server is hosting that protocol, your client config will use the specified protocal based on what is provided as the url:
// nats://localhost:4444
// wss://localhost:4444
// ws://localhost:4444
Conn, err = nats.Connect(url, opts...)
if err != nil {
    panic(fmt.Errorf("error, when connecting to nats service for client init. Error: %v", err))
}

nats server

import (
	"github.com/nats-io/nats-server/v2/server"
)

// When behind proxy make sure the clients configure nats.ProxyPath, see ./nats_client.go
opts := &server.Options{ 
    Port: 3000,
	Authorization: "some-super-secret-token",
	Websocket: server.WebsocketOpts{
		Host:  "0.0.0.0",
		Port:  4430,
		NoTLS: true, // set to true if your terminating tls with proxy, false if not
	},
}
ns, err := server.NewServer(opts)
if err != nil {
    log.Fatalf("error, unable to start nats server. Error: %v", err)
}
go ns.Start()
if !ns.ReadyForConnections(10 * time.Second) {
    log.Fatalf("error, nats failed to start due to timeout")
}

print stack trace

func printStackTrace() {
    // Create a byte slice to hold the stack trace, size 1024 bytes
    buf := make([]byte, 1024)
    // Capture stack trace into buf, runtime.Stack returns the length of n bytes written
    n := runtime.Stack(buf, false)
    // Print the stack trace as a string
    log.Printf("Stack trace:\n%s\n", buf[:n])
}

profiling

stageStart := time.Now()

logStage := func(stage string, stageStart time.Time) {
	log.Printf("%s. Duration: %s\n", stage, time.Since(stageStart))
}

logStage("time spent doing XXXXXXX", stageStart)

// reference: https://en.wikipedia.org/wiki/Orders_of_magnitude_(time)
// nanosecond   ns One billionth of one second 1 ns: The time needed to execute one machine cycle by a 1 GHz microprocessor
// microsecond  μs One millionth of one second 1 μs: The time needed to execute one machine cycle by an Intel 80186 microprocessor
// millisecond  ms One thousandth of one second

Redis get

result, err := config.RedisConnectionPool.Get(ctx, key).Result()
if err != nil {
    if errors.Is(err, redis.Nil) {
        return nil, nil
    }
    return nil, fmt.Errorf("error, when attempting to get key %s from redis. Error %v", key, err)
}

Redis pipeline

rpl := _REDIS_CONNECTION_POOL_.Pool.Pipeline()
var results []model._SOME_STRUCT_
var cmds []*redis.StringCmd

// Prepare the pipeline commands
for _, key := range constants._REDIS_KEY_LIST_ {
	cmds = append(cmds, rpl.Get(ctx, key))
}

// Execute the pipeline
_, err := rpl.Exec(ctx)
if err != nil {
	return nil, fmt.Errorf("error, when executing Redis pipeline: %v", err)
}

// Collect and process results
for i, cmd := range cmds {
	var redisResult string
	redisResult, err = cmd.Result()
	redisValueExists := !errors.Is(err, redis.Nil)
	if err != nil && redisValueExists {
		return nil, fmt.Errorf("error, when fetching value from redis: %v", err)
	}

	var parsedResult *model._SOME_STRUCT_
	parsedResult = &model._SOME_STRUCT_{}
	key := constants._REDIS_KEY_LIST_[i]
	if redisValueExists {
		err = json.Unmarshal([]byte(redisResult), parsedResult)
		if err != nil {
			return nil, fmt.Errorf("error, when attempting to unmarshal redis result. Error: %v", err)
		}
	}
	results = append(results, *parsedResult)
}

return results, nil

Redis set

err := config.RedisConnectionPool.Set(ctx, key, value, expirationDuration).Err()
if err != nil {
    return fmt.Errorf("error, when attempting to set redis key: %s. Error: %v", key, err)
}

Retry

func RequestWithRetry(client *http.Client, request *http.Request, retryLimit int, retryBackoffInSeconds int) (*http.Response, error) {
	var response *http.Response
    retryBackoff := time.Duration(retryBackoffInSeconds) * time.Second
	var err error
	for i := 0; i < retryLimit; i++ {
		response, err = func(client *http.Client, r *http.Request) (*http.Response, error) {
			response, err = client.Do(r)
			if response != nil && (response.StatusCode < 200 || response.StatusCode > 299) {
				if response.StatusCode == http.StatusNotFound {
					log.Printf("recieved a 404 when attempting url: %s", r.URL)
				}
				var rb []byte
				rb, err = io.ReadAll(response.Body)
				if err != nil {
					return nil, fmt.Errorf("error, when reading error response body: %v", err)
				}
				return nil, fmt.Errorf("error, when performing get request. ERROR: %v. RESPONSE CODE: %d. RESPONSE MESSAGE: %s", err, response.StatusCode, string(rb))
			}
			if err != nil {
				if response != nil {
					err = fmt.Errorf("error: %v. RESPONSE CODE: %d", err, response.StatusCode)
				}
				return nil, fmt.Errorf("error, when performing post request. ERROR: %v", err)
			}
			return response, nil
		}(client, request)
		if err != nil {
			err = fmt.Errorf("error, when attempting to send request. Error: %v", err)
			if response != nil {
				err = fmt.Errorf("%v. Response code: %d", err, response.StatusCode)
			}
			log.Printf("error, request failed, retrying in %d seconds", retryBackoffInSeconds)
			time.Sleep(retryBackoff)
		} else {
			break
		}
	}
	if err != nil {
		return nil, err
	}
	return response, nil
}

Single Flight

import (
	"context"
	"fmt"
	"sync"
	"time"

	"golang.org/x/sync/singleflight"
)

type User struct {
	ID   string
	Name string
}

type UserService struct {
	group singleflight.Group
	cache sync.Map
}

func (s *UserService) GetUser(ctx context.Context, userID string) (User, error) {
	if cached, ok := s.cache.Load(userID); ok {
		return cached.(User), nil
	}

	value, err, _ := s.group.Do(userID, func() (any, error) {
		// Double check once we're inside the singleflight call in case
		// another caller filled the cache while we were waiting.
		if cached, ok := s.cache.Load(userID); ok {
			return cached.(User), nil
		}

		user, err := fetchUserFromDB(ctx, userID)
		if err != nil {
			return nil, err
		}

		s.cache.Store(userID, user)
		return user, nil
	})
	if err != nil {
		return User{}, fmt.Errorf("get user %q: %w", userID, err)
	}

	return value.(User), nil
}

func (s *UserService) RefreshUser(ctx context.Context, userID string) (User, bool, error) {
	value, err, shared := s.group.Do(userID, func() (any, error) {
		user, err := fetchUserFromDB(ctx, userID)
		if err != nil {
			return nil, err
		}

		s.cache.Store(userID, user)
		return user, nil
	})
	if err != nil {
		return User{}, false, fmt.Errorf("refresh user %q: %w", userID, err)
	}

	return value.(User), shared, nil
}

func fetchUserFromDB(ctx context.Context, userID string) (User, error) {
	select {
	case <-ctx.Done():
		return User{}, ctx.Err()
	case <-time.After(250 * time.Millisecond):
		fmt.Printf("expensive fetch for user %s\n", userID)
		return User{ID: userID, Name: "Ada"}, nil
	}
}

func exampleSingleflight() {
	service := &UserService{}
	var wg sync.WaitGroup

	for range 5 {
		wg.Add(1)
		go func() {
			defer wg.Done()

			user, shared, err := service.RefreshUser(context.Background(), "user-123")
			if err != nil {
				fmt.Println("error:", err)
				return
			}

			fmt.Printf("user=%+v shared=%t\n", user, shared)
		}()
	}

	wg.Wait()
}

sql migrate

package database

import (
    "strconv"
    "strings"
    "sort"
    "fmt"
    "log"
    "os"
    "regexp"
    "database/sql"
)

func (c *Client) migrate() error {
	initExists, err := c.doesInitTableExist()
	if err != nil {
		return fmt.Errorf("error, when checking if database initialization is needed. Error: %v", err)
	}

	if !initExists {
		log.Println("database is not initialized, creating init table")
		err = c.createInitTable()
		if err != nil {
			return fmt.Errorf("error, when attempting to create init table. Error: %v", err)
		}
		log.Println("database initialization complete")
	}

	log.Println("checking for migrations")
	dirEntries, err := os.ReadDir(c.migrationDir)
	if err != nil {
        return fmt.Errorf("error, when attempting to read database directory. Directory: %s. Error: %v", c.migrationDir, err)
	}
	var migrationFileCandidateFileNames []string
	for _, entry := range dirEntries {
		if !entry.IsDir() {
			migrationFileCandidateFileNames = append(migrationFileCandidateFileNames, entry.Name())
		}
	}

	migrationFiles := c.filterForMigrationFiles(migrationFileCandidateFileNames)
	var migrationsCompleted []string
	noMigrationsToProcessMessage := "no database migration files to process, skipping migrations"
	if len(migrationFiles) == 0 {
		log.Println(noMigrationsToProcessMessage)
		return nil
	} else {
		migrationsCompleted, err = c.checkForCompletedMigrations()
		if err != nil {
			return fmt.Errorf("error, when checking for completed migrations: %v", err)
		}
	}

	migrationsNeeded := c.determineMigrationsNeeded(migrationFiles, migrationsCompleted)
	migrationsNeededSorted := c.sortMigrationsNeededFiles(migrationsNeeded)
	for _, fileName := range migrationsNeededSorted {
		log.Printf("attempting to perform database migration with %s", fileName)

		filePath := fmt.Sprintf("%s/%s", c.migrationDir, fileName)
		err = c.executeSQLFile(filePath)
		if err != nil {
			return fmt.Errorf("error, when executing sql script: Filename: %s. Error: %v", fileName, err)
		}
		err = c.recordSuccessfulMigration(fileName)
		if err != nil {
			return fmt.Errorf("error, when attempting to record a successful migration. Error: %v", err)
		}
	}
	log.Println("finished database schema changes")
	return nil
}

func (c *Client) createInitTable() error {
    _, err := c.conn.Exec(
        `CREATE TABLE init (                                                   
           id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,                    
           migration_file_name TEXT NOT NULL UNIQUE                          
        )`)
    if err != nil {
        return fmt.Errorf("error, when executing query to create init table. Error: %v", err)
    }
	return nil
}

func (c *Client) sortMigrationsNeededFiles(needed []string) []string {
	re := regexp.MustCompile(`^(\d+)`)
	sort.Slice(needed, func(i, j int) bool {
		num1, _ := strconv.Atoi(re.FindStringSubmatch(needed[i])[1])
		num2, _ := strconv.Atoi(re.FindStringSubmatch(needed[j])[1])
		return num1 < num2
	})
	return needed
}

func (c *Client) determineMigrationsNeeded(migrationFiles []string, migrationsCompleted []string) []string {
	var migrationsNeeded []string
	migrationsCompletedMap := make(map[string]bool)
	for _, value := range migrationsCompleted {
		migrationsCompletedMap[value] = true
	}
	for _, value := range migrationFiles {
		if !migrationsCompletedMap[value] {
			migrationsNeeded = append(migrationsNeeded, value)
		}
	}
	return migrationsNeeded
}

func (c *Client) filterForMigrationFiles(candidates []string) []string {
	var migrationFiles []string
	re := regexp.MustCompile(`^\d+`)
	for _, fileName := range candidates {
		if re.MatchString(fileName) {
			migrationFiles = append(migrationFiles, fileName)
		}
	}
	return migrationFiles
}

func (c *Client) recordSuccessfulMigration(fileName string) error {
	_, err := c.conn.Exec(
		`INSERT INTO init (migration_file_name)
        VALUES (?)`,
		fileName,
	)
	if err != nil {
		return fmt.Errorf("error, when attempting to run sql command. Error: %v", err)
	}
	return nil
}

func (c *Client) checkForCompletedMigrations() (results []string, err error) {
	var rows *sql.Rows
	rows, err = c.conn.Query(
		`SELECT migration_file_name
        FROM init`,
	)
	defer func() {
		err = rows.Err()
		if err != nil {
			err = fmt.Errorf("error, when reading rows. Error: %v", err)
		}
		rows.Close()
	}()

	if err != nil {
		return nil, fmt.Errorf("error, when attempting to retrieve pending migrations. Error: %v", err)
	}

	for rows.Next() {
		var result string
		err = rows.Scan(
			&result,
		)
		if err != nil {
			return nil, fmt.Errorf("error, when scanning for pending migrations. Error: %v", err)
		}
		results = append(results, result)
	}

	return results, nil
}

func (c *Client) doesInitTableExist() (bool, error) {
	var result bool
	row := c.conn.QueryRow(
        `SELECT count(*)
        FROM sqlite_master
        WHERE type='table' 
            AND name='init'`,
	)
	err := row.Scan(
		&result,
	)
	if err != nil {
		return false, fmt.Errorf("error, when checking to see if database had been initialized. Error: %v", err)
	}
	return result, nil
}

func (c *Client) executeSQLFile(filePath string) error {
	content, err := os.ReadFile(filePath)
	if err != nil {
		return fmt.Errorf("error, failed to read SQL file. Error: %w", err)
	}

	sql := string(content)
	queries := strings.Split(sql, ";")

	for _, query := range queries {
		query = strings.TrimSpace(query)
		if query == "" {
			continue
		}

		_, err = c.conn.Exec(query)
		if err != nil {
			return fmt.Errorf("error, failed to execute. Query: %s. Error: %v", query, err)
		}
	}

	return nil
}

Task queue

type Task struct {
	Fn     func(chan string)
	Result chan string
}

type TaskQueue struct {
	taskChan chan Task
}

var taskQueueInstance *TaskQueue
var once sync.Once

func GetTaskQueueInstance() *TaskQueue {
	once.Do(func() {
		taskQueueInstance = &TaskQueue{
			taskChan: make(chan Task, 100),
		}
		taskQueueInstance.StartWorkers(10)
	})
	return taskQueueInstance
}

func (q *TaskQueue) StartWorkers(numWorkers int) {
	for i := 0; i < numWorkers; i++ {
		go func() {
			for task := range q.taskChan {
				task.Fn(task.Result)
			}
		}()
	}
}

func (q *TaskQueue) AddTask(task Task) {
	q.taskChan <- task
}

func exampleUsage() {
	// Fetch the singleton instance and start 10 workers
	queue := GetTaskQueueInstance()

	// Create yer own channel to receive the result
	resultChan := make(chan string)

	// Add the task
	queue.AddTask(Task{
		Fn: func(result chan string) {
			result <- "This task be complete, yarrr!"
		},
		Result: resultChan,
	})

	// Wait for result
	result := <-resultChan
	fmt.Println("Result: ", result)

	// Yarrr, add more tasks as ye please!
}

Leader Election Through Postgres

package leader

import (
	"context"
	"errors"

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

const leaderLockID int64 = 123456 // Pick a stable, unique value for this job.

func TryAcquire(ctx context.Context, pool *pgxpool.Pool) (*pgxpool.Conn, bool, error) {
	conn, err := pool.Acquire(ctx)
	if err != nil {
		return nil, false, err
	}

	var acquired bool
	err = conn.QueryRow(ctx,
		`SELECT pg_try_advisory_lock($1)`,
        // SELECT pg_try_advisory_lock(5391437);
        // true  = acquired
        // false = another session owns it
        // does not wait till lock is available

        // SELECT pg_advisory_lock(5391437);
        // waits until available; then returns
		leaderLockID,
	).Scan(&acquired)
	if err != nil {
		conn.Release()
		return nil, false, err
	}

	if !acquired {
		conn.Release()
		return nil, false, nil
	}

	// Keep this conn checked out while this process is leader.
	return conn, true, nil
}

func Release(ctx context.Context, conn *pgxpool.Conn) error {
	var released bool
	err := conn.QueryRow(ctx,
		`SELECT pg_advisory_unlock($1)`,

		leaderLockID,
	).Scan(&released)

	conn.Release()

	if err != nil {
		return err
	}
	if !released {
		return errors.New("leader lock was not held")
	}
	return nil
}

Context Cleanup

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

func main() {
	os.Exit(run())
}

func run() int {
	// Cancel when the process receives Ctrl-C (SIGINT) or a normal stop signal
	// from Docker/Kubernetes/systemd (SIGTERM).
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
	defer signal.Stop(signals) // unregister signal handling / release resources

	appCtx, cancelApp := context.WithCancelCause(context.Background())
	defer cancelApp(nil)

	go func() {
		sig := <-signals
		cancelApp(fmt.Errorf("received shutdown signal: %s", sig))
	}()

	var cleanupWg sync.WaitGroup

	// Your normal application work here
	// before your operations that require cleanup ensure you add cleanupWg.Add(1)
	cleanupWg.Add(1)
	go appStuffDo(appCtx, &cleanupWg, cancelApp)

	cleanupWg.Add(1)
	go otherAppStuffDo(appCtx, &cleanupWg)

	<-appCtx.Done()
	log.Printf("shutting down: %v", context.Cause(appCtx))

	cleanupDone := make(chan struct{})
	go func() {
		cleanupWg.Wait()
		close(cleanupDone)
	}()

	cleanupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	select {
	case <-cleanupDone:
		log.Println("cleanup succeeded")
		return 0

	case <-cleanupCtx.Done():
		log.Println("cleanup deadline exceeded; exiting")
		return 1
	}
}

func appStuffDo(ctx context.Context, wg *sync.WaitGroup, cancelApp context.CancelCauseFunc) {
	// if failure during init then use cancelApp. Init failures should never be ignored otherwise we won't discover a problem until a use-case is triggered

	defer wg.Done()

	// normal work until shutdown
	<-ctx.Done()

	// component-specific cleanup
}

func otherAppStuffDo(ctx context.Context, wg *sync.WaitGroup) {
	defer wg.Done()

	// normal work until shutdown
	<-ctx.Done()

	// component-specific cleanup
}

PostgreSQL array parameters with ANY

Use ANY($1::text[]) when comparing a column against a Go slice:

statuses := []string{"pending", "active"}

rows, err := pool.Query(
      ctx,
      `SELECT id, status
       FROM tasks
       WHERE status = ANY($1::text[])`,
      statuses,
)

This is equivalent to:

WHERE status IN ('pending', 'active')

Key details:

The brackets matter: use text[], not text().

Prefix matching with LIKE

Concatenate a parameter with PostgreSQL’s % wildcard:

prefix := "intro"

rows, err := pool.Query(
      ctx,
      `SELECT id, title
       FROM articles
       WHERE title LIKE $1 || '%'`,
      prefix,
)

This matches titles beginning with intro, such as:

introduction intro to databases introductory guide

Here:

Other common patterns:

-- Ends with the supplied value WHERE title LIKE '%' || $1

-- Contains the supplied value WHERE title LIKE '%' || $1 || '%'

-- Case-insensitive prefix match WHERE title ILIKE $1 || '%'

If the parameter itself can contain % or _, those characters also act as wildcards and may need escaping when literal matching is required.