The guide on how we write backend code.

Our main programming language is Go, and we strive to keep it that way as much as possible, unless there’s a good reason.

This page is more concerning how we generally do stuff in the backend, and not only just Go specifically, but code examples will be in Go.

If and when we introduce a new programming language to our backend (Ehm ehm… python for AI devs), please follow this page very closely and replicate as many of the guardrails and abstractions as you see in the page. Lazily of course, no need to run and implement a pub/sub abstraction if you don’t need to publish / subscribe to events. Also, let’s only use typed languages :)

Go code

The convention we use to write Go code is whatever gofmt says.

In terms of styling of code, we follow the famous Uber’s Go Style Guide. Please take the time to read and internalize it.

To keep things tidy, we validate linting in CI @ .github/workflows/go-lint.yml.

The services run inside our AWS EKS cluster, with the corresponding Helm config files @ infra/helm/apps/vega.

Project structure

In services/go:

  • All dirs other then the ones listed below - Code for the implementation of a service.
  • external - The external API each service exports.
    • For example external/trinomanager/grpc includes the structs and functions generated by the trinomanager’s service .proto.
    • APIs include gRPC, Temporal, HTTP, etc…
  • tests/integration - Code for integration tests.

Database

We use gorm as the ORM to access our Postgresql database.

The code for interacting with the database is in common/db/db.go.

For tenant isolation, we use the GetTenantConnection function which receives a tenant name, and automatically injects WHERE tenant_name = ? to your select queries and sets tenant_name in your insert / update queries. For when you don’t need tenant isolation (for example, when reading all entries of some table), you can use GetRawConnection.

Here’s an example to how we use the DB in our code:

import (
    "vega/common/db"
    "vega/common/models"
)

type Config struct {
    DatabaseConfig db.Config
}

// gorm will name this table example_models.
type ExampleModel struct {
    // Inherit from tenant model to get tenant_name column and the benefits of GetTenantConnection.
    models.TenantModel

    // Always set explicit column name, as some of gorm's API receive a string as an input.
    Example string `gorm:"column:example"`
}

func main() {
    // Read env vars set by the helm file of our service.
    cfg := config.InitializeConfig(Config{})

    // Connect to the db.
    _ = db.InitializeDatabaseFromConfig(&cfg.DatabaseConfig)
}

func OnTenantRequest(ctx context.Context, tenant string) {
    conn := db.GetTenantConnection(ctx, tenant)

    example := ExampleModel{}

    // Notice how gorm returns the error in .Error.
    // This is not picked up by our linter, so please remember to always check .Error.
    err := conn.Find(&example).Error
    if err != nil {
        // ...
    }
}
Transactions in gorm

When you want to execute a transaction in gorm (BEGIN), don’t use the Begin() Rollback() Commit() functions, always use Transaction. It achieves the same thing, but also rollbacks when you return an error or panic!

Separation of models

It’s very important to not put any database models in common and external, only internally in the implementation of the service.

We specifically chose gRPC as our main communication layer, you can learn more about it in Distributed state.

Migrations

We use different migration tools depending on the language:

Go Migrations (Atlas)

We interact with the production database for Go services only through atlas. Atlas is a service that gives you GitOps for database migrations (they call it “Terraform for DBs”).

Let’s say you add a new database model by adding a new struct right below an existing one like ConnectorInstance. Run scripts/migration_gen.sh with option 1 to generate you a new migration file that will be validated in CI by the Migrate Lint workflow, and service/go/dbmigrator will apply the migrations safely in CD. Just don’t forget to add the migrations files in the same PR where you’ve made the relevant changes.

To download and install the version of the Atlas CLI we use, simply run the following in your terminal:

curl -sSf https://atlasgo.sh | ATLAS_VERSION='v0.30.0' sh -s -- -y

# Or with homebrew:
brew install ariga/tap/atlas

Additional packages needed:

  • kubectl
  • yq
Python Migrations (Alembic)

For Python services, we use Alembic for database migrations. Alembic integrates with SQLModel/SQLAlchemy and provides automatic migration generation based on model changes.

To generate a new Python migration:

  1. Add or modify your SQLModel database models in the Python services
  2. Run scripts/migration_gen.sh with option 4 (or option 5 to generate both Go and Python migrations)
  3. The script will automatically discover models from installed packages and generate migration files in services/py/dbmigratorpy/migrations/versions/
  4. The migrations will be validated in CI by the Migrate Lint Python workflow
  5. service/py/dbmigratorpy will apply the migrations safely in CD

The Python migration workflow checks for:

  • Migration conflicts (multiple migrations branching from the same parent)
  • Modified existing migrations (which should never happen)
  • Orphaned migrations (migrations with invalid down_revision references)
  • Duplicate revision IDs

To install Alembic and dependencies, use uv:

cd services/py/dbmigratorpy
uv sync

gRPC

Here’s an example to how you communicate with another service:

import trinoManagerGRPC "vega/external/trinomanager/grpc"

type Config struct {
    TrinoManagerConfig trinoManagerGRPC.Config
}

func main() {
    // Read env vars set by the helm file of our service (for example, gateway is in infra/helm/apps/vega/templates/gateway.yaml).
    cfg := config.InitializeConfig(Config{})

    // Connect to the trinomanager grpc server.
    trinoManagerClient, err = trinoManagerGRPC.InitializeTrinoManagerClient(&cfg.TrinoManagerConfig, "my-service-name")

    // Now we can simply execute rpc calls.
    _ = trinoManagerClient.SyncQuery(ctx, "SELECT * from system.runtime.queries")
}

The .proto & generated gRPC files must reside in external/<my-service-name>, as that is the only directory copied in the service’s Dockerfile.

To keep things simple, you can use scripts/autogen.sh to regenerate all gRPC files when modifying .proto files.

Pub/Sub events

To notify other services that some event has occurred, with no regard to which services are actually interested in the event (decoupled), we use pub/sub over postgres (NOTIFY / LISTEN).

The mechanism is stupidly simple, there’s no queue and no persistency to the messages, if you miss the message, its gone forever.

The code is in common/pubsub/pubsub.go.

Common use case is to get an update on when to resynchronize state. For example, tun2proxy listens to broker creation events (which the gateway notifies when a broker was just created by a user) to re-fetch all brokers.

Here’s how it looks like in the code:

import (
    "time"

    commonDB "vega/common/db"
    "vega/common/pubsub"
)

type Config struct {
    DatabaseConfig commonDB.Config
}

var messagesSubscriber *pubsub.Subscriber[string]
var messagesPublisher pubsub.Publisher[string]

func main() {
    cfg := config.InitializeConfig(Config{})
    db := commonDB.InitializeDatabaseFromConfig(&cfg.DatabaseConfig)
    ctx := context.Background()

    // Notice how you can listen on multiple channels.
	messagesSubscriber, _ = pubsub.InitSubscriber[string](ctx, []string{"channel-name"}, db.GetConnectionURL())
	messagesPublisher = pubsub.InitPublisher[string](db)

    // Sends "hello" every 5 seconds in the background.
    go func() {
        for {
            _ = messagesPublisher.Notify(ctx, "channel-name", "tenant-name", "hello")
            time.Sleep(5 * time.Second)
        }
    }()

    channel, _ := messagesSubscriber.Subscribe("tenant-name")

    // Loop breaks when the channel is closed (call to Unsubscribe).
    for msg := range channel {
        // Will print hello every 5 seconds.
        fmt.Println("%v", msg)
    }
}

GraphQL

GraphQL is a language to create very flexible APIs, we use gqlgen as the server to serve these APIs.

To use graphql, you first create a schema.graphql file, for example:

type Query {
    authors(name: String): [Author!]!
}

type Author {
    name: String!
    books: [Book!]!
}

type Book {
    title: String!
}

And then on the client side you can run the following queries:

# Get all book titles.
{
    authors {
        books {
            title
        }
    }
}

# Get all author names.
{
    authors {
        name
    }
}

# Get all book titles made by J.K. Rowling.
{
    authors(name: "J.K. Rowling") {
        books {
            title
        }
    }
}

As you can see, without changing anything on the server, we are able to pick and choose what data to query. It has many features, you can learn more at the official site.

gqlgen

gqlgen lets you write .graphql files, then generate the Go functions you need to implement, and finally also run the server-side code to serve APIs.

The functions it generates for you to implement are the resolvers. A graphql resolver is run when a query that requests the specific resource. In our example with the authors and books, one such resolver will look like:

import "vega/common/db"

func (r *authorsResolver) Books(ctx context.Context, author *Author) ([]*Book, error) {
    var books []*Book
    err := db.GetTenantConnection(ctx).Find(&books).Where("author_name = ?", author.Name).Error
    if err != nil {
        return nil, err
    }
    return books, nil
}

The generation process is included in scripts/autogen.sh, so after modifying .graphql, just run the script, and implement the resolvers.

Note how the resolver is per author! Because graphql cannot know ahead of time which authors are going to be queried, it runs the resolver function per item in the list. This has major consequences on performance, and is unscalable for queries that request lists of lists of lists… It is exponentially more DB queries for each level you add.

Batching

The solution to that is batching, where you collect the author names that need to be queried, and then query them all after a little while. We’ve developed an abstraction to batching in common/batch/batching.go, here’s how we can improve our books resolver example by using it:

import (
    "vega/common/batch"
    "vega/common/db"
)

const BatchBooksFromAuthors batch.ID = iota

func (r *authorsResolver) Books(ctx context.Context, author *Author) ([]*Book, error) {
    // GetList() returns a list of items.
    // Get() should be used for getting a single item, for example, the latest book written by the author.
	return batch.GetList(
		ctx,

        // The item to add to the batch list.
		author.Name,

        // Execute a query on the batch list, triggers after 2ms or when reaching 1000 items in list.
        // Note that this function is registered for the FIRST item added to the list, so you must not read
        // variables from outside what is received in the trigger function.
        // In our example: You can only use |authorNames| to build the db query.
		func(ctx context.Context, authorNames []string) ([]*Books, error) {
            var books []*Book
            err := db.GetTenantConnection(ctx).Find(&books).Where("author_name IN ?", authorNames).Error
            if err != nil {
                return nil, err
            }
            return books, nil
		},

        // Guide the batch on how to separate the results back.
		func(x *Book) string {
			return x.AuthorName
		},

        // The unique batch ID.
		BatchBooksFromAuthors)
}
Good practices
  • If you return a list of items, make the item inside the list non nullable: [Item!].
  • Separate to different .graphql schema files when adding new types.

Authentication & Authorization

We use gin as our HTTP framework and Descope as our auth provider.

In our gateway service, we add a gin wrapper implemented in gateway/api/wrappers/authentication_wrapper.go to wrap specific HTTP endpoints to require authentication. The wrapper eventually calls Descope’s ValidateSessionWithToken which both authenticates the user’s refresh and session tokens, and returns the user session data, like their assumed roles, which allows for authorization.

Our GraphQL endpoint /query is also able to have more specific authorization rules, using directives (similar to python decorators), for example:

Query {
    dataSource(id: ID): [DataSource!] @authorize(permissions: [DataSourceRead])
    user(email: String): [User!] @authorize(permissions: [UserRead])
}

We manage which permissions are assigned to which role from Descope’s UI.

To add a new permission, head over to Descope and add that permission to the relevant role.

Temporal

We use Temporal when we want to run code that will be guaranteed to finish, even in the case of a crash.

Example use cases:

  • A long running task, where you want to checkpoint steps in the middle, in case of a crash.
  • A distributed transaction where you want to modify state of 2 or more services (SAGA).
  • Run something once every X amount of time (like cron).

To understand how it works, you only need to understand activities & workflows.

Workflows are a blueprint (basically a configuration) of a flow of activities. The Go code you write in a workflow generates the blueprint using code. Theoretically they could’ve just used a declarative language like yaml, but programming languages are much more powerful. This is why workflows must be pure functions, functions that have no side-effects. Examples of side-effects:

  • Writing to a file.
  • Connecting to a socket.
  • Getting a random value.

To actually execute some side-effect, a workflow runs an activity. In the activity is where you actually move forward the execution of a workflow by doing something of value like executing a DB transaction or sending gRPC to another service.

Once an activity is finished, the return value is saved in Temporal’s transaction log, basically creating a checkpoint to the workflow. If the next activity to run crashes, Temporal can resume execution by reading the transaction log. Pretty cool stuff.

Determinism is critical: Everything directly in the workflow (not inside an activity) must be deterministic. When Temporal replays a workflow after a crash, it doesn’t re-execute activities—it remembers their previous responses and ensures calls happen in the same order with the same parameters. This means you cannot use non-deterministic operations directly in workflows. For example, don’t use rand.Int() or time.Now() directly in workflow code. More importantly, you must not trigger child workflows that use AI models (like workflows in the AI service) directly from the workflow, because their responses are non-deterministic. Instead, trigger them from inside an activity. Temporal allow you to set some very useful properties on the execution of an Activity:

ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
    // Timeout to the execution of the activity of 15 minutes, including retries.
    StartToCloseTimeout: time.Minute * 15,

    // Exponential backoff, starting with 1 second between retries, and multiplying by 2 each time, until reaching 1 hour.
    // Once reaching 1 hour, Temporal will retry the activity once an hour, forever.
    RetryPolicy: &temporal.RetryPolicy{
        InitialInterval:        time.Second,
        BackoffCoefficient:     2,
        MaximumInterval:        time.Hour,

        // Do not retry again if the following errors occur.
        NonRetryableErrorTypes: []string{"some-error-we-want-to-abort-immediately-on"},
    },

    // RetryPolicy is optional. If one is not specified a default RetryPolicy is provided by the server.
    // The default RetryPolicy provided by the server specifies:
    // - InitialInterval of 1 second
    // - BackoffCoefficient of 2.0
    // - MaximumInterval of 100 x InitialInterval
    // - MaximumAttempts of 0 (unlimited)
    // To disable retries set MaximumAttempts to 1.
})

var result string
_ = workflow.ExecuteActivity(ctx, SomeActivity, params).Get(ctx, &result)

Please note to always set a StartToCloseTimeout, as the default is unlimited :)

Temporal may look inviting with its colorful UI and playful docs, but you should not use it unless you actually need it, just because it’s so powerful. One simple misconfiguration can cause disasters.

Error Handling

Exposed Errors

When you encounter an error that needs to be exposed to the user, use the exposed error creation functions defined in common/errors/exposed.go. These include:

  • NewExposedError(ctx, code, err)
  • NewExposedErrorf(ctx, code, format, args...)

These functions take the following inputs:

  1. Context: A context.Context object to provide contextual information.
  2. Error Code: An error code from common/errors/errors.pb.go.
  3. Internal Error: An internal error message, not exposed to the user (used for logging).

The user will only see the error code and a corresponding predefined message from common/errors/errors.go. If a new error code is added, ensure you update the .proto file and provide a matching message.

If you need to pass additional arguments with an exposed error, use the Arg method on the error object.

Examples 
import (
    vegaErrors "vega/common/errors"
)

// Basic exposed error
vegaErrors.NewExposedError(ctx, vegaErrors.ErrorCode_INVALID_CONNECTOR_INSTANCE_NAME, err)

// Exposed error with formatting
vegaErrors.NewExposedErrorf(ctx, vegaErrors.ErrorCode_INVALID_CONNECTOR_INSTANCE_NAME, "your message here - %s", message)

// Exposed error with additional arguments
vegaErrors.NewExposedError(ctx, vegaErrors.ErrorCode_INVALID_CONNECTOR_INSTANCE_NAME, err).
    Arg("trino_error", trinoError)

Error Catching and Logging

There are specific locations where errors are caught, exposed to the user, and logged in Datadog. These include:

  • Gin Requests
  • GraphQL Requests
  • Trino Query PubSub

In Datadog, you can filter exposed errors by using the tag UserError:true.

Wrapping External Errors

For functions that return errors from external packages, always wrap the error before returning it. Use the appropriate wrapper based on the context:

  • Temporal Errors (for Temporal activities only):

    • temporal.New*

  • Gin Errors (for Gin requests only):

    • response.NewUnauthorizedError
    • response.NewOK

  • Regular Errors (general use):

    • .Errorf
    • errors.New
    • errors.Unwrap
    • errors.Join
    • .Wrap
    • .Wrapf

Wrapping errors provides better context and ensures proper error handling throughout the application.

Logging & Tracing

We use Datadog for logging, tracing, and monitoring across our services.

In our Go code, we use a custom logger built on top of zerolog. This logger enhances logs by adding additional context, including a VegaLogger flag, which allows us to filter and isolate our logs in Datadog. This is crucial because logs from external packages can generate noise, and we want only our logs visible in Datadog.

To ensure seamless tracing, we use interceptors wherever possible. These interceptors propagate trace IDs automatically. To correlate logs with traces, you should use the Get function from common/logger/logger.go. This function attaches the trace ID stored in the context.Context to the logger.

It is essential always to provide a context.Context wherever possible. For example, you should use http.NewRequestWithContext instead of http.NewRequest. The only scenario where you might not have a context is in the main function. Since main has no inherited context like HTTP, gRPC, or Temporal requests, you must create a new context. Additionally, because our custom logger adds metadata, always use it in these cases.

In Temporal workflows, you do not have access to context.Context. For these cases, use the GetWorkflowLogger function from common/logger/logger.go.

If you want to rate-limit specific logs, such as logging only once every 5 seconds, you can use the GetRateLimitted function. This helps reduce noise while ensuring critical logs are still captured.

Example:

import (
    "time"
    "vega/common/logger"
)

// Keep the object, as it manages the state for rate limitting.
// The convention is to name it 'l'.
l := logger.GetRateLimitted(ctx, 5 * time.Second)

l.Warn().Msg("hello world")

Kubernetes & Helm

Helm is a tool for templating k8s configuration files. We use ArgoCD to monitor changes to Helm charts, which are rendered into k8s manifests that are applied to the k8s cluster.

The templating syntax used is Go’s text/template package.

Open infra/helm/apps/vega/templates/gateway.yaml, there you should see how we configure the:

  • Deployment - Sets the docker image and environment variables.
  • Service - Sets exposed ports.
  • HorizontalPodAutoscaler - Sets automatic scaling of pods (both up & down). We usually set CPU and memory monitoring, but you can get creative here like monitoring some queue size that your service pops items from.

External / internal enforcement

The Dockerfiles for the services are in infra/docker/go, let’s look at infra/docker/go/gateway/Dockerfile to get a taste:

FROM 860592944260.dkr.ecr.us-east-2.amazonaws.com/library/golang:1.24-alpine

WORKDIR /app

COPY go.mod go.sum ./

RUN go mod download

COPY gateway ./gateway/
COPY common ./common/
COPY external ./external/

WORKDIR /app/gateway
RUN go build -o /gateway

ENTRYPOINT ["/gateway"]

As you can see, the service copies only 3 directories:

  • gateway - The directory of the internal implementation.
  • common - Our common go package.
  • external - The external APIs to all our other services.

This is how we enforce that there is no import to some other service’s internal implementation code.

To understand why we go to great lengths to keep this separations, read How to deal with eventual consistency.

Integration Tests

Integration tests are the sweetspot between E2E (end to end) tests and unit tests. While unit tests are fine, they break often as the implementation changes.

E2E tests are good, and we use them, but they are slow to run, slow to develop, error prone, and statistic by nature.

An integration test is like a unit test, but instead of mocking interfaces and services, it has a setup step that runs whatever dependency the test requires. We use Tilt to run all our services, and then we run the integration tests.

Go testing

Our integration tests look & behave the same as regular Go unit tests (just interacting with the cluster run by Tilt).

There are two ways to assert in a test:

  • require - Stops the test immediately.
  • assert - Continues the test.

Both fail the test and print out the assertion failure message, but you should always think which is more appropriate. If you don’t know, use require.

When logging, use t.Logf, never zerolog or our own logging infra. The reason for this is that this allows the output of the tests to look as if they were run sequentially (one after another), even though we use t.Parallel() to run most of our tests in parallel.

We use gotestsum to run the tests in CI / CD. This tool’s output is nicer than go test (making the output of each test collapsable in github actions), and outputs a junit.xml file, which is a known test run summary file format and can be used by other tools to generate beautiful UI.

To run integration tests locally:

# Cluster needs to be up and running.
tilt up prod

# Working dir needs to be services/go.
cd services/go

# Run all tenant isolation tests using a regex filter.
# Can also remove the -run flag to run all tests.
# -count=1 means to always run the tests, and not check the cached results of last run.
go test ./tests/integration -v -run '^TestTrinoTenantIsolation.*$' -count=1

Tilt & Local debugging

To setup your local environment to work with k8s over Tilt, run scripts/setup_kubernetes.sh -s. This will install kind on your machine.

As a prerequisite to running tilt, run ./scripts/login_ecr.py to login to AWS ECR and create the needed secret for the k8s inside kind to be able to pull from ECR.

Then whenever you want to run the services do tilt up local in the monorepo root dir. Tilt serves a web UI on http://localhost:10350, where you can see the progress and liveness of the pods running on your local k8s.

Notice the profile given is local (instead of prod as in the previous example), the difference between them is that prod will pull the latest main images from ECR (cached in the local values.versions.tilt.local.yaml file) and local builds all the services (including trino) from your local code. See tilt/profile.bzl’s PROFILES variable and how it configures these profiles and more.

If you want to quickly run as prod with just one of the services as local, instead of creating a new profile, you can also run the following (example for gateway): tilt up prod -- --override gateway=local

Tilt is configured using the Tiltfile in the monorepo root dir. When running using local, it is configured so whenever you save changes to some file, it will update the relevant services inside the running k8s cluster. Do note that it will only set the service in “changed” state and show a cyan background behind the reload icon, so only once you click it, it will reset to the new version in k8s.

Sometimes we notice the pods not really updating to the latest code after clicking the cyan reload icon, what we do in these cases is click the reload icon once more…

To debug with breakpoints, use local-debug instead of local in your profile. For example: tilt up prod -- --override gateway=local-debug.

Then inside your favorite editor configure dlv (the go debugger) to attach to port 400xx (the port for the specific service can be seen in the service’s config in tilt/config.bzl).

Please note that debug for some reason needs at least 200Mi memory, so remember to modify the service’s values.tilt.yaml to have more memory if you get CrashloopBackoff after setting local-debug.

Taskfile

As part of the efforts to improve the local environments, we use a tool called Taskfile https://taskfile.dev/ - to add some standardization of how to run utilities locally and in the CI.
What’s special about our use of it is that all commands are running in containers and running exactly the same locally and in the CI, without needing any special installations on your local env.

Please install task cli either from here or run sh -c "$(curl --location https://taskfile.dev/install.sh)" -- -d

To see which commands are available you can always run task --list

> ./bin/task
task: Available tasks for this project:
* app:autogen:                  Generate all files from sources, like proto & mocks
* app:helm-template-test:       Test all helm templates are valid
* go:lint:                      Lint all golang files
* go:unit-tests:                Run all golang unit-tests
* setup:aws-sso-login:          Run the aws-sso-login script. append args with --
* setup:login-ecr-tilt:         Login to AWS and ECR in Tilt
* setup:refresh-login:          Refresh all login credentials for aws, teleport and docker