cloudrift-go

Cloud-agnostic abstraction for storage, messaging, document databases, cache, secrets, pub/sub, and email — built for Lyzr microservices. The Go port of cloudrift (Python).

Context-first. Every operation takes a context.Context and is backed by native Go cloud SDKs (aws-sdk-go-v2, azure-sdk-for-go, mongo-driver/v2, go-redis/v9) — connection-pooled, no wrappers.
Drop-in providers. Same interface across AWS, Azure, and self-hosted backends. Swap s3 ↔ azure_blob (or sqs ↔ azure_bus, documentdb ↔ cosmos, redis ↔ elasticache ↔ azure_redis, ses ↔ azure_acs ↔ smtp) by changing one string.
Multiple auth methods per provider. Static keys, IAM roles, profiles, managed identity, service principals, SAS tokens, mTLS, IAM auth — pick what your microservice already has.

Category	Factory	AWS	Azure	Self-hosted
Storage	`storage.New`	`s3`	`azure_blob`	—
Messaging	`messaging.New`	`sqs`	`azure_bus`	—
Document DB	`document.New`	`documentdb`	`cosmos`	—
Cache	`cache.New`	`elasticache`	`azure_redis`	`redis`
Secrets	`secrets.New`	`aws_secrets_manager`	`azure_keyvault`	`env` / `file` / `memory`
Pub/Sub	`pubsub.New`	`sns`	`azure_eventgrid`	—
Email	`email.New`	`ses`	`azure_acs`	`smtp`

Install

go get github.com/LYZR-OSS/cloudrift-go

Go 1.23+. Import only the categories you use — Go links only what you import, so a service using just the cache never pulls AWS/Azure SDK code into its binary.

Quick start

Every backend is constructed via a factory function and held for the lifetime of the service. Reuse one instance per resource — the underlying client is connection-pooled.

import "github.com/LYZR-OSS/cloudrift-go/storage"

// Construct once at startup
st, err := storage.New(ctx, "s3", storage.Config{
    Bucket:             "my-bucket",
    AWSAccessKeyID:     "AKIA...",
    AWSSecretAccessKey: "...",
    Region:             "us-east-1",
})
if err != nil { ... }

// Use anywhere
_, err = st.Upload(ctx, "docs/hello.txt", []byte("hello world"), "text/plain")
data, err := st.Download(ctx, "docs/hello.txt")
url, err := st.PresignedURL(ctx, "docs/hello.txt", time.Hour)

// Release sockets at shutdown
defer st.Close(ctx)

Configuration via env vars

Pick the provider per environment with a single env var:

st, err := storage.New(ctx, os.Getenv("STORAGE_PROVIDER"), storage.Config{ // "s3" in prod, "azure_blob" in dev
    Bucket:    os.Getenv("STORAGE_BUCKET"),
    Container: os.Getenv("STORAGE_CONTAINER"),
    Region:    os.Getenv("STORAGE_REGION"),
    // ...
})

The factory routes to the right auth method based on which credential fields are set (mirroring the Python library's kwargs-presence routing); unset fields are simply ignored.

Storage

// AWS S3
storage.New(ctx, "s3", storage.Config{Bucket: "b", Region: "us-east-1"})       // IAM role
storage.New(ctx, "s3", storage.Config{Bucket: "b", AWSAccessKeyID: "...",      // static keys
    AWSSecretAccessKey: "...", Region: "us-east-1"})
storage.New(ctx, "s3", storage.Config{Bucket: "b", ProfileName: "dev"})        // ~/.aws/credentials

// Azure Blob
storage.New(ctx, "azure_blob", storage.Config{ConnectionString: "...", Container: "c"})
storage.New(ctx, "azure_blob", storage.Config{AccountURL: "https://acct.blob.core.windows.net",
    AccountKey: "...", Container: "c"})
storage.New(ctx, "azure_blob", storage.Config{AccountURL: "...", SASToken: "...", Container: "c"})
storage.New(ctx, "azure_blob", storage.Config{AccountURL: "...", Container: "c"}) // managed identity
storage.New(ctx, "azure_blob", storage.Config{AccountURL: "...", Container: "c",
    TenantID: "...", ClientID: "...", ClientSecret: "..."})                       // service principal

Operations — same on every backend:

key, err := st.Upload(ctx, key, data, "application/json")
data, err := st.Download(ctx, key)
err = st.Delete(ctx, key)
ok, err := st.Exists(ctx, key)
keys, err := st.List(ctx, "logs/")
for key, err := range st.ListIter(ctx, "logs/") { ... } // lazy, true pagination
url, err := st.PresignedURL(ctx, key, time.Hour)
key, err = st.Copy(ctx, src, dst)
md, err := st.GetMetadata(ctx, key)
key, err = st.UploadStream(ctx, key, reader, "video/mp4")
err = st.Close(ctx)

Azure note: PresignedURL requires account-key auth (connection string or AccountKey) — SAS URLs cannot be minted from managed-identity credentials.

Messaging

import "github.com/LYZR-OSS/cloudrift-go/messaging"

// AWS SQS
q, err := messaging.New(ctx, "sqs", messaging.Config{
    QueueURL: "https://sqs.us-east-1.amazonaws.com/.../q", Region: "us-east-1"})

// Azure Service Bus
q, err := messaging.New(ctx, "azure_bus", messaging.Config{
    ConnectionString: "...", QueueName: "my-queue"})
q, err := messaging.New(ctx, "azure_bus", messaging.Config{
    FullyQualifiedNamespace: "ns.servicebus.windows.net", QueueName: "my-queue"}) // managed identity

Operations:

Bodies are raw bytes (binary-safe — JSON, OTLP protobuf, base64, anything), with optional string attributes (SQS MessageAttributes / Service Bus ApplicationProperties):

// Raw bytes + attributes
id, err := q.Send(ctx, []byte("…otlp bytes…"), map[string]string{"encoding": "otlp_proto"}, 0)
ids, err := q.SendBatch(ctx, []messaging.OutgoingMessage{
    {Body: []byte("a"), Attributes: map[string]string{"k": "v"}},
    {Body: []byte("b")},
})

// JSON convenience for object payloads
id, err = messaging.SendJSON(ctx, q, map[string]any{"action": "process", "id": 42}, nil, 0)

msgs, err := q.Receive(ctx, 10, 20*time.Second) // long-poll
for _, m := range msgs {
    handleJob(m.Body)            // []byte; m.Attributes is map[string]string
    obj, _ := m.JSON()           // decode body as a JSON object when applicable
    err = q.Delete(ctx, m.ReceiptHandle)              // ack
    // or: q.DeadLetter(ctx, m.ReceiptHandle, "bad payload")
}

depth, err := q.GetQueueDepth(ctx)
err = q.Purge(ctx)
err = q.Close(ctx)

For cross-account access, set Config.RoleARN (+ optional ExternalID) on messaging.New/storage.New — the AWS client assumes the role via STS on top of the base credentials.

Dead-lettering: Azure Service Bus dead-letters natively. SQS has no per-message dead-letter API, so the backend emulates it: it re-sends the message body to the DLQ (from Config.DLQURL or the queue's RedrivePolicy) and deletes the original.

Document Database

document.New is a connection factory: it returns a configured *mongo.Client from the official MongoDB Go driver regardless of provider — both AWS DocumentDB and Azure Cosmos DB (MongoDB API) speak the MongoDB wire protocol. You get the driver's full native API: typed decoding, transactions, bulk writes, change streams, aggregation.

import "github.com/LYZR-OSS/cloudrift-go/document"

// AWS DocumentDB (MongoDB-compatible)
client, err := document.New("documentdb", document.Config{
    URI:       "mongodb://user:pass@cluster.docdb.amazonaws.com:27017/?tls=true",
    TLSCAFile: "/etc/ssl/rds-ca-bundle.pem",
    MaxPoolSize: 200,
})
client, err := document.New("documentdb", document.Config{ // or host/port credentials
    Host: "cluster.docdb.amazonaws.com", Port: 27017, Username: "u", Password: "p"})
client, err := document.New("documentdb", document.Config{ // or mTLS
    Host: "cluster.docdb.amazonaws.com", Port: 27017, Username: "u", Password: "p",
    TLSCertKeyFile: "/etc/ssl/client.pem"})

// Azure Cosmos DB (MongoDB API)
client, err := document.New("cosmos", document.Config{ConnectionString: "mongodb://..."})
client, err := document.New("cosmos", document.Config{Account: "myacct", AccountKey: "..."})

Operations — the native driver API, identical on both providers:

import "go.mongodb.org/mongo-driver/v2/bson"

coll := client.Database("lyzr").Collection("users")

res, err := coll.InsertOne(ctx, bson.M{"name": "Alice", "age": 30})
err = coll.FindOne(ctx, bson.M{"name": "Alice"}).Decode(&user)
cur, err := coll.Find(ctx, bson.M{"age": bson.M{"$gte": 18}})
n, err := coll.UpdateOne(ctx, bson.M{"_id": res.InsertedID},
    bson.M{"$set": bson.M{"age": 31}})

err = client.Disconnect(ctx) // lifecycle is caller-managed; call at shutdown

Cosmos note: key-based auth only (connection string or account + key) — Cosmos for MongoDB (RU) does not accept Azure AD tokens at the wire-protocol layer. Account + AccountKey builds the URI with the Cosmos-required parameters (ssl, replicaSet=globaldb, retryWrites=false).

Cache

import "github.com/LYZR-OSS/cloudrift-go/cache"

// Self-hosted Redis
c, err := cache.New(ctx, "redis", "from_url", cache.Config{URL: "redis://localhost:6379/0"})
c, err := cache.New(ctx, "redis", "from_credentials", cache.Config{
    Host: "redis.internal", Port: 6379, Password: "...", DB: 0})

// AWS ElastiCache
c, err := cache.New(ctx, "elasticache", "from_auth_token", cache.Config{
    Host: "my-cluster.cache.amazonaws.com", AuthToken: "..."})
c, err := cache.New(ctx, "elasticache", "from_iam_auth", cache.Config{
    Host: "my-cluster.cache.amazonaws.com",
    Username: "lyzr-app", Region: "us-east-1"}) // SigV4 + auto-refresh

// Azure Cache for Redis
c, err := cache.New(ctx, "azure_redis", "from_access_key", cache.Config{
    Host: "my-cache.redis.cache.windows.net", AccessKey: "..."})
c, err := cache.New(ctx, "azure_redis", "from_managed_identity", cache.Config{
    Host: "my-cache.redis.cache.windows.net", Username: "lyzr-app"})

Operations — KV, hash, set, list, counters:

err = c.Set(ctx, "session:abc", data, time.Hour)
val, err := c.Get(ctx, "session:abc") // nil, nil if missing
n, err := c.Delete(ctx, "session:abc")

n, err = c.HSet(ctx, "user:1", "name", "Alice")
fields, err := c.HGetAll(ctx, "user:1")

added, err := c.SAdd(ctx, "dau:2026-06-10", userID) // returns newly-added count (dedup signal)
common, err := c.SInter(ctx, "dau:2026-06-09", "dau:2026-06-10")

n, err = c.LPush(ctx, "jobs", "job-1", "job-2")
batch, err := c.LRange(ctx, "jobs", 0, 99)

n, err = c.Incr(ctx, "hits:home")

// Atomic multi-command round trip (MULTI/EXEC)
err = c.Pipeline(ctx, func(p cache.Pipeliner) {
    p.SAdd("dau:2026-06-10", userID)
    p.Expire("dau:2026-06-10", 48*time.Hour)
})

err = c.Close(ctx)

Secrets

import "github.com/LYZR-OSS/cloudrift-go/secrets"

sm, err := secrets.New(ctx, "aws_secrets_manager", secrets.Config{Region: "us-east-1"}) // IAM
kv, err := secrets.New(ctx, "azure_keyvault", secrets.Config{
    VaultURL: "https://myvault.vault.azure.net"}) // managed identity

// Non-cloud backends — same interface, no secrets manager required (dev, on-prem, CI, tests):
env, err := secrets.New(ctx, "env", secrets.Config{Prefix: "SECRET_"})        // read SECRET_<name> env vars
file, err := secrets.New(ctx, "file", secrets.Config{FilePath: "/run/secrets.json"}) // JSON {name: value}
mem, err := secrets.New(ctx, "memory", secrets.Config{                         // in-memory (alias "local")
    Mapping: map[string]string{"db-password": "s3cret"}})

val, err := sm.GetSecret(ctx, "db-password")
cfg, err := sm.GetSecretJSON(ctx, "db-config")
err = sm.SetSecret(ctx, "db-password", "s3cret")
err = sm.DeleteSecret(ctx, "db-password")
names, err := sm.ListSecrets(ctx, "db-")

Pub/Sub

import "github.com/LYZR-OSS/cloudrift-go/pubsub"

ps, err := pubsub.New(ctx, "sns", pubsub.Config{Region: "us-east-1"})
ps, err := pubsub.New(ctx, "azure_eventgrid", pubsub.Config{
    Endpoint: "https://topic.region.eventgrid.azure.net/api/events", AccessKey: "..."})

id, err := ps.Publish(ctx, topicARN, `{"event": "user.created"}`,
    map[string]string{"source": "auth-svc"})
ids, err := ps.PublishBatch(ctx, topicARN, []pubsub.BatchMessage{
    {Message: `{"n": 1}`}, {Message: `{"n": 2}`},
}) // SNS chunks at the 10-entry batch limit automatically

Email

import "github.com/LYZR-OSS/cloudrift-go/email"

// AWS SES (SESv2)
em, err := email.New(ctx, "ses", email.Config{
    Region: "us-east-1", DefaultFrom: "noreply@example.com"})                  // IAM role / env
em, err := email.New(ctx, "ses", email.Config{AWSAccessKeyID: "AKIA...",       // static keys
    AWSSecretAccessKey: "...", Region: "us-east-1", DefaultFrom: "noreply@example.com"})
em, err := email.New(ctx, "ses", email.Config{ProfileName: "dev",              // ~/.aws profile
    Region: "us-east-1", DefaultFrom: "noreply@example.com"})

// Azure Communication Services (talks to the ACS REST API; no Go SDK needed)
em, err := email.New(ctx, "azure_acs", email.Config{                           // access-key signing
    ConnectionString: "endpoint=https://x.communication.azure.com;accesskey=...",
    DefaultFrom: "DoNotReply@example.com"})
em, err := email.New(ctx, "azure_acs", email.Config{                           // managed identity
    Endpoint: "https://x.communication.azure.com", DefaultFrom: "DoNotReply@example.com"})
em, err := email.New(ctx, "azure_acs", email.Config{                           // service principal
    Endpoint: "https://x.communication.azure.com", TenantID: "...",
    ClientID: "...", ClientSecret: "...", DefaultFrom: "DoNotReply@example.com"})

// Raw SMTP (SendGrid, Mailgun, Postmark, Office365, MailHog, ...)
em, err := email.New(ctx, "smtp", email.Config{Host: "smtp.sendgrid.net",      // STARTTLS, port 587 (default)
    Username: "apikey", Password: "...", DefaultFrom: "noreply@example.com"})
em, err := email.New(ctx, "smtp", email.Config{Mode: "tls", Host: "smtp.example.com", // implicit TLS
    Port: 465, Username: "user", Password: "pw", DefaultFrom: "noreply@example.com"})
em, err := email.New(ctx, "smtp", email.Config{Mode: "plaintext",              // MailHog / Mailpit (dev)
    Host: "localhost", Port: 1025, DefaultFrom: "noreply@example.test"})

Operations — same on every backend. Build an EmailMessage; From falls back to Config.DefaultFrom when empty, and at least one of BodyText / BodyHTML must be set:

id, err := em.Send(ctx, email.EmailMessage{
    To:       []string{"alice@example.com"},
    Subject:  "Welcome",
    BodyText: "Plain text body",
    BodyHTML: "<p>HTML body</p>",
    Cc:       []string{"bob@example.com"},
    Bcc:      []string{"audit@example.com"}, // envelope-only, never in headers
    ReplyTo:  []string{"support@example.com"},
    Attachments: []email.Attachment{{
        Filename: "welcome.pdf", Content: pdfBytes, ContentType: "application/pdf"}},
    Headers: map[string]string{"X-Campaign": "welcome-v2"},
})

ids, err := em.SendBatch(ctx, []email.EmailMessage{ // loops Send by default
    {To: []string{"alice@example.com"}, Subject: "hi",  BodyText: "hi"},
    {To: []string{"bob@example.com"},   Subject: "hi2", BodyHTML: "<b>hi2</b>"},
})

ok := em.HealthCheck(ctx)
err = em.Close(ctx)

Default sender. Each backend takes a DefaultFrom in its config; calls that leave EmailMessage.From empty fall back to it. SES requires the sender (address or domain) to be verified; ACS requires the sending domain to be linked to the resource.

Errors

All backends translate provider-native errors into one hierarchy under core, built on wrapped sentinel errors — match with errors.Is at any level:

import "github.com/LYZR-OSS/cloudrift-go/core"

data, err := st.Download(ctx, "missing.txt")
switch {
case errors.Is(err, core.ErrObjectNotFound):  // specific
case errors.Is(err, core.ErrStorage):         // any storage error
case errors.Is(err, core.ErrCloudRift):       // any cloudrift error
}

Specific errors per category: ErrObjectNotFound/ErrStoragePermission, ErrQueueNotFound/ErrMessageSend, ErrDocumentConnection, ErrCacheConnection/ErrCacheKeyNotFound, ErrSecretNotFound/ErrSecretPermission, ErrTopicNotFound/ErrPublish, ErrEmailSend/ErrRecipientRejected/ErrSenderUnverified/ErrEmailThrottled. Operations a provider cannot honor return core.ErrNotImplemented. The document package returns a native *mongo.Client, so operation errors come from the MongoDB driver directly.

Connection pooling & lifecycle

Every backend holds one long-lived client reused across all operations:

Don't call storage.New(...) inside a request handler.
Do construct it once at startup and share it (struct field, DI container, or package-level singleton).
Release sockets at shutdown with backend.Close(ctx). (AWS/Azure HTTP-based backends are no-op closes; Redis, Mongo, and Service Bus actually tear down connections.)

Testing

go test ./...

The cache suite runs against an in-process miniredis (the Go analogue of the Python suite's fakeredis), so no real cloud credentials are needed. AWS backends accept Config.EndpointURL for LocalStack/MinIO-style integration testing.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Repository files navigation

cloudrift-go

Install

Quick start

Configuration via env vars

Storage

Messaging

Document Database

Cache

Secrets

Pub/Sub

Email

Errors

Connection pooling & lifecycle

Testing

About

Uh oh!

Releases 3

Packages

Uh oh!

Contributors

Uh oh!

Languages

Name		Name	Last commit message	Last commit date
Latest commit History 13 Commits
.github/workflows		.github/workflows
cache		cache
core		core
document		document
email		email
messaging		messaging
pubsub		pubsub
secrets		secrets
storage		storage
.gitignore		.gitignore
README.md		README.md
VERSION		VERSION
go.mod		go.mod
go.sum		go.sum

Uh oh!

Folders and files

Latest commit

History

Repository files navigation

cloudrift-go

Install

Quick start

Configuration via env vars

Storage

Messaging

Document Database

Cache

Secrets

Pub/Sub

Email

Errors

Connection pooling & lifecycle

Testing

About

Resources

Uh oh!

Stars

Watchers

Forks

Releases 3

Packages 0

Uh oh!

Contributors

Uh oh!

Languages

Packages