Help us shape the future of Go! Take the 2026 Go Developer Survey

blobkit

package module
v0.0.0-...-753ccf0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 10, 2026 License: MIT Imports: 17 Imported by: 0

README ¶

BlobKit

Go Reference Go Report Card Go Version

BlobKit is a modular, high-performance object storage library for Go. Engineered with 8 native, production-hardened storage engines with zero external SDK bloat, it provides a clean, unified storage layer for applications interfacing with AWS S3, Cloudflare R2, MinIO, Wasabi, Backblaze B2, Azure Blob Storage, Google Cloud Storage (Native JSON API), WebDAV (Nextcloud / TrueNAS), Google Drive (Shared Drives), Local POSIX Filesystem, Remote SFTP / SSH, and Ephemeral In-Memory storage.

BlobKit offers two distinct operating modes depending on your architecture:

  1. Low-Level Core (Bucket): A fast, stateless, key-value storage abstraction with direct streaming (io.WriteCloser), zero database dependencies, and sub-microsecond throughput.
  2. High-Level Orchestrator (Client): A full-featured storage engine providing canonical object identities (UUIDv7), database metadata registries, policy verification, intelligent multi-cloud routing (circuit breakers, failover), and automated lifecycle management.

Table of Contents

Architectural Principles

BlobKit enforces a strict 3-tier boundary:

+-----------------------------------------------------------------------------------+
| 1. APPLICATION LAYER (Microservices, Web Backends, Workers)                      |
|    - Declares semantic intent: Namespace, OwnerID, Original Filename, Visibility |
|    - Streams raw data (io.Reader) with predictable memory usage                   |
|    - Queries objects by metadata attributes without knowing raw bucket keys       |
+-----------------------------------------------------------------------------------+
                                         |
                                         v
+-----------------------------------------------------------------------------------+
| 2. BLOBKIT INFRASTRUCTURE LAYER (Orchestration Engine)                            |
|    - Canonical ObjectID generation (UUIDv7) & structured partition keys           |
|    - Zero-rewind 512B MIME sniffing & streaming SHA-256 integrity verification     |
|    - Security Policy Engine (MIME/extension whitelist, path & CRLF sanitization)  |
|    - Resumable multipart upload session lifecycle & state tracking                |
|    - Retention policies, legal hold enforcement & decoupled background sweeper    |
|    - Advanced Routing (Weighted traffic, 3-state Circuit Breaker, Capabilities)   |
|    - Bounded LRU caching with negative caching & vendor-neutral telemetry         |
|    - Database Metadata Registry coordination (System of Record)                   |
+-----------------------------------------------------------------------------------+
                                         |
                                         v
+-----------------------------------------------------------------------------------+
| 3. PROVIDER DRIVER LAYER (Wire Abstraction)                                       |
|    - AWS SDK v2 Driver (AWS S3, Cloudflare R2, MinIO, Wasabi, Backblaze B2)       |
|    - Native Azure Blob Storage Driver (Block Blobs, SAS, Azurite emulator)        |
|    - Native Google Cloud Storage Driver (JSON API v1, 256KB Resumable, V4 Sign)   |
|    - WebDAV Protocol Driver (Nextcloud, ownCloud, TrueNAS, Apache/Nginx)          |
|    - Google Drive Driver (Personal & Enterprise Shared Drives, appProperties)     |
|    - Local POSIX Filesystem Driver (Atomic swaps, Sidecar metadata, Edge)         |
|    - Remote SFTP / SSH Driver (Connection pooling, Remote seeks, Sidecars)        |
|    - Pure In-Memory Driver for offline unit tests and ephemeral workloads         |
+-----------------------------------------------------------------------------------+
                                         |
                                         v
+-----------------------------------------------------------------------------------+
| 4. PHYSICAL STORAGE BACKENDS                                                      |
|    AWS S3 / R2 / MinIO / Azure Blob / GCS / WebDAV / GDrive / Local NVMe / SFTP   |
+-----------------------------------------------------------------------------------+
Application Layer Anti-Patterns Prevented by BlobKit
Anti-Pattern BlobKit Solution
Handcrafting raw S3 keys (user/123/avatar.png) Handled by structured, collision-proof Key Generators
Hardcoding bucket names across application code Centralized Driver configuration and Namespace Routers
Storing full CDN/S3 URLs in PostgreSQL Dynamically resolved via ResolveURL or PresignGet
Buffering entire file payloads in RAM to compute SHA-256 Streaming verification via io.TeeReader in $O(1)$ memory
Blindly trusting user-supplied Content-Type headers Zero-rewind 512B magic signature sniffing
Exposing internal proxy IPs (127.0.0.1) in client errors Automated regex scrubber stripping credentials and dials

Identity & Delivery Separation

BlobKit strictly decouples four distinct concepts:

Concept Representation Example Canonical DB Field?
Logical Identity ObjectID 01925b3a-7f28-7102-8f92-9428ad0e451b (UUIDv7) Yes (Primary Key)
Physical Storage Key ObjectKey media/uploads/2026/10/06/01925b3a...png Yes (Storage Index)
Client Presentation Original Filename annual_report_2026.pdf Yes (Header/Audit)
Delivery URL Access URL https://cdn.example.com/... or presigned URL NO (Generated on-demand)

[!IMPORTANT] URLs are transient delivery details, never canonical identifiers. Storing full URLs in databases causes severe data corruption when domains migrate, buckets change, or certificates rotate. In BlobKit, URLs are resolved dynamically via ResolveURL or PresignGet.

Feature Matrix

Feature BlobKit Implementation Guarantee
Lifecycle & Compliance Retention policies, WORM locking, Legal hold, Soft delete, Restore, Permanent purge Thread-safe, atomic state checks
Background GC Decoupled lifecycle.Sweeper (expired objects, soft-delete grace period, stale uploads) Standalone or embedded worker
Object Operations Native server-side Copy, Cross-driver stream copy, Move, Rename, Exists, Head Zero byte movement for renames
Integrity Streaming SHA-256 via io.TeeReader, client digest validation, byte-count verification $O(1)$ memory consumption
Resumable Multipart State machine (Active, Committed, Aborted), per-part upload, restartable sessions Concurrency & failure resilient
Security & Policy MIME/Extension whitelisting, spoofing detection, path traversal & CRLF injection filter Hard boundary defense
Multi-Cloud Routing 3-State Circuit Breaker, Weighted traffic, Active-Passive Failover, Namespace dispatch Zero allocation circuit check
Caching Layer Bounded thread-safe LRU, Negative caching for 404 stampede protection, Mutex invalidation Sub-microsecond cache hits
Observability Vendor-neutral Observer interface, thread-safe MetricsCollector, zero secret leaks OpenTelemetry / Prometheus ready
Developer Experience Typed errors (IsNotFound, IsObjectLocked), testutil.NewMockStorage test fixture Zero cloud credentials for tests

Installation

go get github.com/suhwr/blobkit

Choosing Your API Tier: Low-Level Bucket vs. High-Level Client

BlobKit is architected to give developers full flexibility. Choose the layer that matches your system requirements:

Capability Low-Level Core (Bucket) High-Level Orchestrator (Client)
Primary Identity Direct Physical Key ("reports/2026.pdf") Canonical ObjectID (UUIDv7) + Key
Database Dependency None (Zero Database, Stateless) Pluggable Metadata Registry (Postgres, SQLite, Memory)
Object State Machine Direct I/O (Put, Get, Delete) States: Pending âž” Committed âž” Deleted
Streaming Primitives io.WriteCloser via bucket.NewWriter() Streaming with MIME sniff & SHA-256 verification
Routing & Topologies Single Driver / Bucket Multi-driver Failover, Circuit Breaking & Tiering
Lifecycle & Compliance Provider-native lifecycle Soft-delete, WORM legal hold, automated background sweeper
Best For Microservices, background workers, pure storage I/O Central storage platforms, multi-tenant SaaS, audit systems
🚀 Quick Start: Low-Level Bucket (Stateless Direct I/O)

Ideal for microservices or tasks where you simply want universal storage without running a database:

package main

import (
    "context"
    "fmt"
    "io"
    "strings"

    "github.com/suhwr/blobkit"
    "github.com/suhwr/blobkit/provider/s3"
)

func main() {
    ctx := context.Background()

    // 1. Initialize any storage driver
    driver, _ := s3.NewDriver(s3.Config{
        Region:          "us-east-1",
        Bucket:          "my-app-uploads",
        AccessKeyID:     "MY_KEY",
        SecretAccessKey: "MY_SECRET",
    })

    // 2. Wrap into a stateless Bucket (no database required!)
    bucket, _ := blobkit.NewBucket(driver)
    defer bucket.Close()

    // 3. Put an object stream
    key := "docs/invoice_001.txt"
    _, _ = bucket.Put(ctx, key, strings.NewReader("Invoice Total: $120.00"), blobkit.PutOptions{
        ContentType: "text/plain",
    })

    // 4. Convenience helpers: PutBytes & GetBytes
    _ = bucket.PutBytes(ctx, "cache/flag.bin", []byte{0x01, 0x02}, blobkit.PutOptions{})
    data, _, _ := bucket.GetBytes(ctx, "cache/flag.bin", blobkit.GetOptions{})
    fmt.Printf("Read %d bytes\n", len(data))

    // 5. Streaming io.WriteCloser (zero memory buffering)
    writer, _ := bucket.NewWriter(ctx, "logs/app.log", blobkit.PutOptions{
        ContentType: "text/plain",
    })
    writer.Write([]byte("Application booted successfully\n"))
    writer.Close() // Automatically finalizes upload to S3

    // 6. Inspect & Delete
    exists, _ := bucket.Exists(ctx, key)
    if exists {
        _ = bucket.Delete(ctx, key)
    }
}

Low-Level Storage Engine & Wire-Level Architecture

The Low-Level Core in BlobKit (blobkit.Bucket) is engineered for maximum throughput, predictable memory footprint, and architectural minimalism. When building microservices, background ingestion workers, data pipelines, high-performance reverse proxies, or edge runtimes, you don't need a relational database, UUIDv7 object identity mappings, or complex lifecycle state machines. You simply need fast, reliable, zero-overhead storage I/O directly on the wire.

+-----------------------------------------------------------------------------+
|                         APPLICATION CODE / PIPELINE                         |
|   (io.Reader / io.Writer / io.ReaderAt / io.ReadSeeker / []byte payloads)   |
+-----------------------------------------------------------------------------+
                                      |
              +-----------------------+-----------------------+
              |                                               |
              v                                               v
+-------------------------------+             +-------------------------------+
|      blobkit.Bucket           |             |      SeekableReader           |
|  - Direct Key-Value Storage   |             |  - io.ReadSeekCloser          |
|  - Asynchronous NewWriter()   |             |  - io.ReaderAt (Concurrency)  |
|  - Fast PutBytes / GetBytes   |             |  - HTTP Byte-Range Seeks      |
|  - Low-level Multipart SPI    |             |  - ZIP / Parquet / Media      |
+-------------------------------+             +-------------------------------+
              |                                               |
              +-----------------------+-----------------------+
                                      |
                                      v
+-----------------------------------------------------------------------------+
|                         DRIVER INTERCEPTOR PIPELINE                         |
|   WrapDriver(driver, LatencyMiddleware, CustomRateLimiter, AuditLogger)     |
+-----------------------------------------------------------------------------+
                                      |
                                      v
+-----------------------------------------------------------------------------+
|                     NORMATIVE DRIVER CONTRACT & SPI                         |
|   blobkit.Driver interface  <--- embedded by --->  blobkit.BaseDriver       |
+-----------------------------------------------------------------------------+
                                      |
                                      v
+-----------------------------------------------------------------------------+
|                     ZERO-COPY KERNEL STREAMING                              |
|   ObjectReader.WriteTo(w) -> Linux sendfile(2) / splice(2) fast path        |
+-----------------------------------------------------------------------------+
1. Stateless Bucket Architecture & Direct Key I/O

A Bucket wraps any blobkit.Driver directly. It communicates using physical storage keys without touching an external database:

  • Ultra-low latency: Reads resolve in ~735 ns/op and uploads in ~2.3 µs/op (over 3.3x faster than full client orchestration).
  • Direct key addressing: Store and retrieve using exact paths (e.g. media/2026/avatar.png).
  • Memory-efficient helpers: In-memory byte slices can be saved and retrieved via PutBytes and GetBytes without manual buffer management.
  • Direct provider multipart: Initiate, upload parts, and complete provider multipart sessions directly using raw keys (CreateMultipart, UploadPart, CompleteMultipart).
package main

import (
    "context"
    "fmt"
    "strings"

    "github.com/suhwr/blobkit"
    "github.com/suhwr/blobkit/provider/s3"
)

func main() {
    ctx := context.Background()

    // 1. Initialize any storage driver
    driver, err := s3.NewDriver(s3.Config{
        Region:          "us-east-1",
        Bucket:          "my-app-uploads",
        AccessKeyID:     "MY_KEY",
        SecretAccessKey: "MY_SECRET",
    })
    if err != nil {
        panic(err)
    }

    // 2. Wrap into a stateless Bucket (no database required!)
    bucket, err := blobkit.NewBucket(driver)
    if err != nil {
        panic(err)
    }
    defer bucket.Close()

    // 3. Put an object stream directly to a key
    key := "docs/invoice_001.txt"
    _, err = bucket.Put(ctx, key, strings.NewReader("Invoice Total: $120.00"), blobkit.PutOptions{
        ContentType: "text/plain",
        Metadata: map[string]string{
            "tier": "enterprise",
        },
    })

    // 4. Convenience fast paths: PutBytes & GetBytes
    _ = bucket.PutBytes(ctx, "cache/flags.bin", []byte{0x01, 0x02, 0x03}, blobkit.PutOptions{})
    data, meta, _ := bucket.GetBytes(ctx, "cache/flags.bin", blobkit.GetOptions{})
    fmt.Printf("Read %d bytes (content-type: %s)\n", len(data), meta.ContentType)

    // 5. Inspect metadata (Head) & Existence check
    exists, _ := bucket.Exists(ctx, key)
    if exists {
        head, _ := bucket.Head(ctx, key)
        fmt.Printf("Key: %s, Size: %d bytes\n", head.Key, head.Size)
    }

    // 6. Direct server-side Copy & Delete
    _ = bucket.Copy(ctx, key, "docs/invoice_001_backup.txt")
    _ = bucket.Delete(ctx, key)

    // 7. Paginated listing by key prefix
    result, _ := bucket.List(ctx, blobkit.ListOptions{
        Prefix: "docs/",
        Limit:  50,
    })
    for _, obj := range result.Objects {
        fmt.Printf("- %s (%d bytes)\n", obj.Key, obj.Size)
    }
}
2. Asynchronous Streaming Writes (bucket.NewWriter)

When streaming dynamically generated content (e.g. database dumps, telemetry streams, audio encodes, or compressed archives), buffering the entire output in memory or temporary disk files wastes memory and adds disk I/O latency.

bucket.NewWriter(ctx, key, opts) returns a standard io.WriteCloser backed by an internal asynchronous pipe (io.Pipe). As bytes are written into the writer, they are streamed concurrently to the remote storage driver. Calling .Close() automatically flushes the pipe and awaits provider confirmation:

package main

import (
    "compress/gzip"
    "context"
    "log"

    "github.com/suhwr/blobkit"
)

// Stream dynamically compressed gzip data directly to object storage with O(1) memory
func StreamArchive(ctx context.Context, bucket *blobkit.Bucket, key string) error {
    // 1. Open an asynchronous streaming writer
    writer, err := bucket.NewWriter(ctx, key, blobkit.PutOptions{
        ContentType: "application/gzip",
    })
    if err != nil {
        return err
    }

    // 2. Wrap writer in standard streaming compressor
    gz := gzip.NewWriter(writer)

    // 3. Write data chunks on-the-fly
    payload := []byte("streamed log entry line 1\nstreamed log entry line 2\n")
    if _, err := gz.Write(payload); err != nil {
        // Abort the pipe immediately on error to terminate the background upload
        _ = writer.CloseWithError(err)
        return err
    }

    // 4. Flush and close gzip framing blocks first
    if err := gz.Close(); err != nil {
        _ = writer.CloseWithError(err)
        return err
    }

    // 5. Close the BucketWriter to finalize upload and receive remote persistence error if any
    if err := writer.Close(); err != nil {
        log.Printf("Upload failed on storage provider: %v", err)
        return err
    }

    log.Println("Streaming upload successfully committed")
    return nil
}
3. Kernel Zero-Copy Streaming (ObjectReader.WriteTo)

BlobKit's ObjectReader implements standard Go io.WriterTo:

func (r *ObjectReader) WriteTo(w io.Writer) (int64, error)

When serving objects over HTTP handlers, reverse proxies, or TCP connections (*net.TCPConn), Go runtime's io.Copy detects io.WriterTo. If the underlying driver stream is backed by an OS file descriptor or socket, the transfer engages Linux kernel zero-copy mechanisms like sendfile(2) or splice(2). This moves data straight across kernel buffers without allocating heap memory or copying bytes into userspace Go application memory:

package main

import (
    "fmt"
    "net/http"

    "github.com/suhwr/blobkit"
)

// High-throughput HTTP Asset Server leveraging kernel zero-copy transfer
func BlobServerHandler(bucket *blobkit.Bucket) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        key := r.URL.Path[1:] // e.g. "assets/video.mp4"

        // Open object stream
        reader, err := bucket.Get(r.Context(), key, blobkit.GetOptions{})
        if err != nil {
            if blobkit.IsNotFound(err) {
                http.NotFound(w, r)
                return
            }
            http.Error(w, "storage error", http.StatusInternalServerError)
            return
        }
        defer reader.Close()

        // Set response metadata headers
        if reader.ContentType != "" {
            w.Header().Set("Content-Type", reader.ContentType)
        }
        if reader.Size > 0 {
            w.Header().Set("Content-Length", fmt.Sprintf("%d", reader.Size))
        }

        // Fast-Path: WriteTo streams directly to http.ResponseWriter
        // Utilizing sendfile(2)/splice(2) fast paths with zero heap allocations
        _, _ = reader.WriteTo(w)
    }
}
4. Random-Access Seeking (SeekableReader & zip.NewReader)

Cloud object storage (AWS S3, Cloudflare R2, Google Cloud Storage, Azure Blob Storage) operates on HTTP byte ranges (Range: bytes=start-end), whereas Go standard library packages (such as archive/zip.NewReader, Parquet table readers, video atom decoders, or PDF parsers) require io.ReaderAt or io.ReadSeekCloser.

bucket.OpenSeeker(ctx, key) creates an adapter that bridges this gap seamlessly:

  • Implements io.ReadSeekCloser and io.ReaderAt.
  • Thread-safe ReadAt calls execute range GET requests concurrently across goroutines.
  • Reuses active body streams when sequential reads continue from current cursor.
Example: Extracting a File from a 50GB Remote ZIP Without Downloading It

In the ZIP file specification, the Central Directory index is located at the very end of the archive. A standard downloader would need to download all 50GB. With SeekableReader, Go's zip.NewReader only fetches the trailing metadata bytes over Range GET, then jumps directly to the compressed offset of the requested file:

package main

import (
    "archive/zip"
    "context"
    "fmt"
    "io"

    "github.com/suhwr/blobkit"
)

func ExtractZipEntry(ctx context.Context, bucket *blobkit.Bucket, zipKey string, targetFile string) ([]byte, error) {
    // 1. Open random-access seeker over remote cloud blob
    seeker, err := bucket.OpenSeeker(ctx, zipKey)
    if err != nil {
        return nil, err
    }
    defer seeker.Close()

    // 2. Mount directly into archive/zip.NewReader
    // Only reads the end-of-archive Central Directory bytes over HTTP Range!
    zipReader, err := zip.NewReader(seeker, seeker.Size())
    if err != nil {
        return nil, fmt.Errorf("failed to parse zip central directory: %w", err)
    }

    // 3. Locate and stream only the specific file
    for _, file := range zipReader.File {
        if file.Name == targetFile {
            rc, err := file.Open()
            if err != nil {
                return nil, err
            }
            defer rc.Close()

            // Stream and read ONLY the target file bytes!
            return io.ReadAll(rc)
        }
    }

    return nil, fmt.Errorf("entry %q not found in remote archive", targetFile)
}
5. Building Custom Storage Drivers (BaseDriver SPI)

BlobKit is completely extensible. You can write your own storage driver (e.g. for Redis, IPFS, Ceph RADOS, Couchbase, or internal corporate storage) in under 50 lines of Go code by embedding blobkit.BaseDriver.

Why Embed blobkit.BaseDriver?

The normative BlobKit Driver Contract specifies that any capability not supported by a driver must return blobkit.ErrUnsupportedOperation. Embedding BaseDriver provides safe, contract-compliant stub implementations for all optional methods (Copy, List, PresignGet, CreateMultipart, etc.), so you only need to implement the core methods your backend supports:

package mydriver

import (
    "bytes"
    "context"
    "io"
    "sync"

    "github.com/suhwr/blobkit"
)

// CustomDriver implements blobkit.Driver for an in-memory key-value engine.
type CustomDriver struct {
    blobkit.BaseDriver // Provides compliant stubs for all unadvertised capabilities
    data map[string][]byte
    mu   sync.RWMutex
}

func NewCustomDriver(name string) *CustomDriver {
    return &CustomDriver{
        BaseDriver: blobkit.NewBaseDriver(
            name,
            blobkit.CapCore|blobkit.CapByteRangeGet, // Advertised capability bitmask
        ),
        data: make(map[string][]byte),
    }
}

// 1. Put stores a blob stream
func (d *CustomDriver) Put(ctx context.Context, obj *blobkit.Object, r io.Reader, opts blobkit.PutOptions) (*blobkit.Object, error) {
    payload, err := io.ReadAll(r)
    if err != nil {
        return nil, blobkit.WrapError("put", obj.Key, d.Name(), err)
    }

    d.mu.Lock()
    d.data[obj.Key] = payload
    d.mu.Unlock()

    res := *obj
    res.Size = int64(len(payload))
    return &res, nil
}

// 2. Get retrieves a blob stream
func (d *CustomDriver) Get(ctx context.Context, key string, opts blobkit.GetOptions) (*blobkit.ObjectReader, error) {
    d.mu.RLock()
    payload, exists := d.data[key]
    d.mu.RUnlock()

    if !exists {
        return nil, blobkit.WrapError("get", key, d.Name(), blobkit.ErrObjectNotFound)
    }

    return &blobkit.ObjectReader{
        Object: blobkit.Object{Key: key, Size: int64(len(payload))},
        Body:   io.NopCloser(bytes.NewReader(payload)),
    }, nil
}

// 3. Head inspects blob metadata
func (d *CustomDriver) Head(ctx context.Context, key string) (*blobkit.Object, error) {
    d.mu.RLock()
    payload, exists := d.data[key]
    d.mu.RUnlock()

    if !exists {
        return nil, blobkit.WrapError("head", key, d.Name(), blobkit.ErrObjectNotFound)
    }
    return &blobkit.Object{Key: key, Size: int64(len(payload))}, nil
}

// 4. Delete removes a blob
func (d *CustomDriver) Delete(ctx context.Context, key string) error {
    d.mu.Lock()
    delete(d.data, key)
    d.mu.Unlock()
    return nil
}

// 5. Close releases driver resources
func (d *CustomDriver) Close() error {
    return nil
}
Verifying Custom Drivers with the Contract Test Suite

You can verify your custom driver against BlobKit's normative contract suite in a single line:

package mydriver_test

import (
    "testing"
    "github.com/suhwr/blobkit"
    "github.com/suhwr/blobkit/testutil"
)

func TestCustomDriverContract(t *testing.T) {
    testutil.RunDriverContractTests(t, func() blobkit.Driver {
        return NewCustomDriver("test-custom")
    })
}
6. Driver Middleware & Interceptor Pipeline

BlobKit provides a composable middleware architecture inspired by HTTP middleware. Middlewares intercept wire-level driver operations to inject observability, chaos testing, audit trails, encryption, or rate limiting.

Architecture: WrapDriver & DelegateDriver
  • blobkit.WrapDriver(driver, middlewares...): Composes multiple middlewares outside-in.
  • blobkit.DelegateDriver: A transparent proxy struct that implements blobkit.Driver. Middleware authors embed DelegateDriver and selectively override only the methods they want to intercept.
package main

import (
    "context"
    "log"
    "time"

    "github.com/suhwr/blobkit"
    "github.com/suhwr/blobkit/provider/s3"
)

// Example 1: Built-in Latency Middleware (Measures exact wire duration)
func withLatencyInstrumentation(rawDriver blobkit.Driver) blobkit.Driver {
    return blobkit.WrapDriver(rawDriver, blobkit.LatencyMiddleware(func(op blobkit.Operation, key string, duration time.Duration, err error) {
        log.Printf("[METRICS] Op=%s Key=%s Latency=%v Err=%v", op, key, duration, err)
    }))
}

// Example 2: Custom Security Audit Middleware
type SecurityAuditMiddleware struct {
    blobkit.DelegateDriver
    logger *log.Logger
}

func NewSecurityAuditMiddleware(logger *log.Logger) blobkit.DriverMiddleware {
    return func(next blobkit.Driver) blobkit.Driver {
        return &SecurityAuditMiddleware{
            DelegateDriver: blobkit.NewDelegateDriver(next),
            logger:         logger,
        }
    }
}

// Intercept only Delete operations
func (m *SecurityAuditMiddleware) Delete(ctx context.Context, key string) error {
    m.logger.Printf("[SECURITY AUDIT] Permanent delete requested for key=%s", key)
    return m.DelegateDriver.Delete(ctx, key)
}

func main() {
    rawDriver, _ := s3.NewDriver(s3.Config{
        Region: "us-east-1",
        Bucket: "app-bucket",
    })

    // Chain middlewares: Audit runs first, then Latency, then the raw Driver
    instrumentedDriver := blobkit.WrapDriver(
        rawDriver,
        blobkit.LatencyMiddleware(func(op blobkit.Operation, key string, duration time.Duration, err error) {
            log.Printf("Op %s took %v", op, duration)
        }),
        NewSecurityAuditMiddleware(log.Default()),
    )

    bucket, _ := blobkit.NewBucket(instrumentedDriver)
    // All bucket operations pass through the middleware chain
}
7. Low-Level Error Taxonomy & Resilient Retries

BlobKit standardizes all errors across cloud providers into a unified taxonomy wrapped in blobkit.StorageError:

type StorageError struct {
    Op       string // "get", "put", "head", "delete", etc.
    Key      string // Object key that triggered the error
    Provider string // Provider identifier (e.g. "s3-primary", "azure-archive")
    Err      error  // Underlying sentinel error
}
Error Classification Predicates
Helper Function Indicates Safe to Retry? Example Sentinel Errors
blobkit.IsNotFound(err) Object or bucket does not exist No ErrObjectNotFound, ErrBucketNotFound
blobkit.IsPermanent(err) Non-recoverable client errors No ErrInvalidKey, ErrPermissionDenied, ErrChecksumMismatch, ErrSecurityViolation
blobkit.IsTransient(err) Temporary network / server outages Yes ErrProviderUnavailable, ErrRateLimited, 502/503/504, connection reset
blobkit.IsSecurityViolation(err) Security / MIME policy blocked No ErrSecurityViolation, ErrMIMEMismatch
blobkit.IsObjectLocked(err) Object locked by WORM / legal hold No ErrObjectLocked
blobkit.IsRateLimited(err) Throttling / 429 Too Many Requests Yes (with backoff) ErrRateLimited
Resilient Retry Loop Example
package main

import (
    "context"
    "errors"
    "log"
    "time"

    "github.com/suhwr/blobkit"
)

func FetchWithRetry(ctx context.Context, bucket *blobkit.Bucket, key string) ([]byte, error) {
    maxRetries := 3
    backoff := 150 * time.Millisecond

    for attempt := 1; attempt <= maxRetries; attempt++ {
        data, _, err := bucket.GetBytes(ctx, key, blobkit.GetOptions{})
        if err == nil {
            return data, nil
        }

        // 1. Immediately abort on permanent non-retryable errors
        if blobkit.IsPermanent(err) {
            if blobkit.IsNotFound(err) {
                log.Printf("Object not found: %s", key)
            } else if blobkit.IsPermissionDenied(err) {
                log.Printf("Access forbidden: check IAM permissions")
            }
            return nil, err
        }

        // 2. Retry only transient recoverable errors
        if blobkit.IsTransient(err) {
            log.Printf("Attempt %d transiently failed (%v). Retrying in %v...", attempt, err, backoff)
            select {
            case <-ctx.Done():
                return nil, ctx.Err()
            case <-time.After(backoff):
                backoff *= 2 // Exponential backoff
                continue
            }
        }

        return nil, err
    }

    return nil, errors.New("maximum retries exceeded")
}
Automated Credential Scrubbing

To prevent accidental data leakage into log ingestion services (Sentry, Datadog, CloudWatch), all errors returned by BlobKit automatically scrub:

  • AWS, Azure, and Google Cloud HMAC signatures (X-Amz-Signature, X-Goog-Signature, sig=...)
  • Bearer tokens, Basic Auth headers, and OAuth refresh tokens
  • RSA and Ed25519 private key PEM headers
  • Internal RFC 1918 IP addresses (10.x.x.x, 172.16-31.x.x, 192.168.x.x, 127.0.0.1)
8. Wire-Level Protocols & Concurrency Internals Deep Dive

BlobKit interfaces directly with storage backends at the raw wire level without heavy, monolithic vendor SDKs. Understanding how BlobKit manages chunking, concurrency, atomic state transitions, and memory allocations allows you to build mission-critical infrastructure with deterministic behavior.

a. Provider Wire Protocols for Resumable & Chunked Uploads

Each storage provider implements its own chunking specification. BlobKit unifies these protocols behind the normative Driver interface:

Provider Protocol Architecture Wire Mechanics & Finalization Guarantees & Edge Cases
AWS S3 / R2 / MinIO AWS Multipart Upload POST /?uploads $\rightarrow$ PUT /?partNumber=N&uploadId=ID $\rightarrow$ POST /?uploadId=ID (XML block list) MD5 / SHA-256 etag verification per part. Same-key copy uses MetadataDirectiveReplace.
Azure Blob Storage Azure Block Blobs PUT /blob?comp=block&blockid=ID (base64) $\rightarrow$ PUT /blob?comp=blocklist (XML block list) Preserves stored.Size on unknown-size streams. Appends SAS tokens to x-ms-copy-source on internal Copy.
Google Cloud Storage GCS Resumable Upload POST /upload/storage/v1/b/... (session URI) $\rightarrow$ PUT <session_uri> with Content-Range: bytes START-END/TOTAL Aborts session on client context cancellation or size mismatch (ErrSizeMismatch).
Google Drive Resumable Drive v3 API POST /upload/drive/v3/files?uploadType=resumable $\rightarrow$ PUT <uri> with 256 KiB chunks $\rightarrow$ bytes */uploaded finalizer Ref-counted per-key mutex locks prevent duplicate files on race conditions. Explicit boundary finalization prevents 308 hang.
Local POSIX Filesystem Staged Temporary Files Chunks written to tmp_<uploadID> $\rightarrow$ atomic rename os.Rename(tmp, dst) + sidecar .meta.json Atomic rename ensures zero partial reads. Suffix range queries (bytes=-N) parse from EOF.
Remote SFTP / SSH Remote Atomic Staging Staged via sftp.Client $\rightarrow$ atomic remote rename $\rightarrow$ sidecar .meta.json Non-destructive atomic move first, avoiding premature deletion of existing target files.
b. Thread Safety & Race-Free Concurrency Model

BlobKit is verified with the Go race detector (go test -race ./...) under high concurrent workloads:

  1. Per-Session Lock Serialization: Concurrent calls to client.UploadPart targeting the same resumable session are serialized through sessionLocks sync.Map with per-session mutexes. This prevents lost updates where two concurrent worker goroutines simultaneously fetch and overwrite the session's uploaded parts slice.

  2. Stampede Defense with Panic-Safe Singleflight: The singleflightGroup collapses concurrent identical Head or metadata queries into a single driver lookup. If the leader goroutine panics, the panic is caught, the error is safely propagated to all awaiting follower goroutines, and the leader re-panics—preventing deadlocks, memory leaks, or followers receiving phantom (nil, nil) results.

  3. Immutable Deep Cloning: Metadata maps and session parts retrieved from or saved into in-memory caches (cache/lru.go) and registries (registry/memory.go) are deeply cloned. External modifications to returned structs will never mutate internal cache entries or cause concurrent map read/write panics.

  4. Non-Destructive Rollback Compensation: In client.Copy, destination pre-existence is audited prior to initiating the copy. If a post-copy metadata save fails, BlobKit's rollback compensation only deletes the destination file if it did not exist prior to the operation. Pre-existing files are never deleted on rollback.

c. Kernel Zero-Copy & RFC 9110 Byte-Range Streaming
  • Linux Zero-Copy (sendfile / splice): BlobKit's ObjectReader exposes WriteTo(w io.Writer). When streaming to a network socket (such as an http.ResponseWriter backed by net.TCPConn), Go's internal runtime engages sendfile(2) or splice(2). Bytes stream directly from file descriptors or driver sockets to the network card without touching application user-space memory buffers.
  • RFC 9110 Range Compliance: All drivers declaring CapByteRangeGet strictly parse HTTP Range headers, including:
    • Standard ranges: bytes=0-1024 (inclusive offsets 0 through 1024)
    • Open-ended ranges: bytes=2048- (offset 2048 through EOF)
    • Suffix ranges: bytes=-500 (trailing 500 bytes of the object)
    • Exact EOF behavior: SeekableReader.ReadAt returns io.EOF whenever bytes read are less than the requested buffer at object boundaries, guaranteeing 100% compliance with Go's standard library io.ReaderAt contracts.

High-Level Orchestrator & Enterprise Features

1. Multi-Cloud Provider Configurations

BlobKit natively supports 8 production-grade storage engines and declarative fleet management. For in-depth architecture, configuration options, memory bounds, and security practices, refer to the dedicated provider guides:

Provider Package Supported Backends / Protocols Dedicated Guide
AWS S3 & Compatible provider/s3 AWS S3, Cloudflare R2, MinIO, Wasabi, Backblaze B2 Read S3 Guide ↗
Azure Blob Storage provider/azure Native Azure Block Blobs, SAS Presigning, Azurite emulator Read Azure Guide ↗
Google Cloud Storage provider/gcs Native GCS JSON API v1, 256KB Resumable, V4 Signed URLs Read GCS Guide ↗
WebDAV Protocol provider/webdav Nextcloud, ownCloud, TrueNAS, Apache/Nginx, Synology/QNAP Read WebDAV Guide ↗
Google Drive provider/gdrive Google Drive v3, Personal & Enterprise Shared Drives Read Google Drive Guide ↗
Local POSIX Filesystem provider/fs NVMe, SSD, EBS, Edge Storage, Atomic Swaps, Sidecar Meta Read FS Guide ↗
Remote SFTP / SSH provider/sftp Remote Linux/Unix Servers, SSH Keys, Connection Pooling Read SFTP Guide ↗
In-Memory Harness provider/memory Pure RAM Ephemeral, Zero-Credential Unit Testing Read Memory Guide ↗
Multi-Cloud Fleet router/fleet Declarative JSON/YAML Multi-Cloud Fleet & Topologies Read Fleet Guide ↗
a. Cloudflare R2 (Zero Egress CDN)
package main

import (
    "github.com/suhwr/blobkit/provider/s3"
)

func createR2Driver() (*s3.Driver, error) {
    return s3.NewDriver(s3.Config{
        Name:            "r2-primary",
        Endpoint:        "https://<account_id>.r2.cloudflarestorage.com",
        Bucket:          "production-assets",
        AccessKeyID:     "YOUR_R2_ACCESS_KEY_ID",
        SecretAccessKey: "YOUR_R2_SECRET_ACCESS_KEY",
        PublicBaseURL:   "https://cdn.example.com", // Custom Cloudflare CDN domain
        MaxConcurrentStreams: 32,                 // Worker semaphore limit
        MaxMemoryBytes:       128 * 1024 * 1024,  // 128MB RAM budget for multipart buffers
    })
}
b. AWS S3 (Standard / Multi-Region)
func createS3Driver() (*s3.Driver, error) {
    return s3.NewDriver(s3.Config{
        Name:            "aws-s3-backup",
        Region:          "us-east-1",
        Bucket:          "production-backup-vault",
        AccessKeyID:     "YOUR_AWS_ACCESS_KEY_ID",
        SecretAccessKey: "YOUR_AWS_SECRET_ACCESS_KEY",
        UsePathStyle:    false, // Virtual hosted-style addressing for S3
    })
}
c. MinIO & Self-Hosted S3
func createMinIODriver() (*s3.Driver, error) {
    return s3.NewDriver(s3.Config{
        Name:            "local-minio",
        Endpoint:        "http://127.0.0.1:9000",
        Bucket:          "local-dev-bucket",
        AccessKeyID:     "minioadmin",
        SecretAccessKey: "minioadmin",
        UsePathStyle:    true, // Required for local MinIO / path-style clusters
    })
}
d. In-Memory Driver (Offline Testing & CI/CD)
import "github.com/suhwr/blobkit/provider/memory"

// Pure in-memory driver: zero network calls, instant execution, race-tested
memDriver := memory.NewDriver(memory.Config{
    Name:          "in-memory",
    Bucket:        "test-bucket",
    PublicBaseURL: "https://test.example.com",
})
e. Google Drive Driver (Cloud & Shared Drives)

Native Google Drive REST API v3 implementation adhering to pure object storage semantics with resumable chunked uploads, byte-range streaming, in-memory key indexing, and Shared Drive support:

import "github.com/suhwr/blobkit/provider/gdrive"

gdriveDriver, err := gdrive.NewDriver(gdrive.Config{
    Name:              "gdrive-primary",
    FolderID:          "1A2B3C4D5E6F7G8H9I0J", // Target Google Drive folder or Shared Drive ID
    SupportsAllDrives: true,                   // Enable Google Workspace Shared Drives
    TokenFunc: func(ctx context.Context) (string, error) {
        // Return fresh OAuth2 token or use Service Account JWT
        return oauthTokenSource.Token(ctx)
    },
    ChunkSize:        8 * 1024 * 1024, // 8 MiB chunks (must be multiple of 256 KiB)
    KeyCacheCapacity: 10000,           // Concurrency-safe LRU key -> fileID cache
})
f. Local POSIX Filesystem Driver (Zero Cost & Edge Storage)

Microsecond-latency, zero-cost persistent disk storage with atomic temporary file commits, byte-range seeking (os.File.Seek), sidecar metadata persistence, and strict path traversal protection:

import "github.com/suhwr/blobkit/provider/fs"

fsDriver, err := fs.NewDriver(fs.Config{
    Name:              "fs-local",
    RootDir:           "/var/data/blobs",               // Dedicated directory path
    PublicBaseURL:     "https://cdn.example.com/blobs", // Optional public URL prefix
    DirMode:           0755,
    FileMode:          0644,
    EnableSidecarMeta: true,                            // Atomically persist .meta.json sidecars
    StagingDir:        ".staging",                      // Chunk staging directory for multipart
})
g. WebDAV Protocol Driver (Nextcloud, ownCloud, TrueNAS, NAS)

Native RFC 4918 implementation connecting BlobKit to self-hosted cloud platforms (Nextcloud, ownCloud), TrueNAS (ZFS pools), Apache/Nginx WebDAV, and private NAS appliances with streaming PUT/GET, byte-range seeks, and automatic MKCOL collection provisioning:

import "github.com/suhwr/blobkit/provider/webdav"

webdavDriver, err := webdav.NewDriver(webdav.Config{
    Name:          "nextcloud-primary",
    Endpoint:      "https://cloud.example.com/remote.php/dav/files/admin",
    Username:      "admin",
    Password:      "app-password-or-token",
    PublicBaseURL: "https://cloud.example.com/s", // Optional public link prefix
})
h. Azure Blob Storage Driver (Native Block Blobs & SAS Presigning)

Native Azure Blob Storage driver adhering to BlobKit's pure object storage abstraction. Supports Block Blob uploads, server-side block staging (comp=block) and commits (comp=blocklist), SharedKey HMAC-SHA256 authentication, Service SAS token generation for presigned URLs, byte-range streaming, server-side synchronous copy, batch deletes, and credential scrubbing:

import "github.com/suhwr/blobkit/provider/azure"

azureDriver, err := azure.NewDriver(azure.Config{
    Name:           "azure-primary",
    AccountName:    "mystorageaccount",
    AccountKey:     "base64-encoded-storage-key",
    ContainerName:  "my-container",
    // Optional: custom endpoint for Azurite emulator or sovereign clouds
    // CustomEndpoint: "http://127.0.0.1:10000/mystorageaccount",
    PublicBaseURL:  "https://cdn.example.com", // Optional public CDN prefix
})
i. Google Cloud Storage Driver (Native JSON API & V4 Signed URLs)

Native Google Cloud Storage driver interfacing directly with the GCS JSON API v1 and Resumable Upload protocol. Supports single-shot media uploads, multipart/related uploads with custom metadata, 256 KiB-aligned chunked resumable sessions, byte-range streaming, server-side rewrite/copy, concurrent batch deletion, and client-side Google Cloud V4 Signed URLs (GOOG4-RSA-SHA256) with zero external Google Cloud SDK dependencies:

import "github.com/suhwr/blobkit/provider/gcs"

gcsDriver, err := gcs.NewDriver(gcs.Config{
    Name:                "gcs-primary",
    Bucket:              "my-cloud-bucket",
    TokenFunc: func(ctx context.Context) (string, error) {
        // Return fresh OAuth2 token or use Compute Engine / Workload Identity token source
        return tokenSource.Token(ctx)
    },
    // Optional: for Google Cloud V4 Signed URLs (PresignGet & PresignPut)
    ServiceAccountEmail: "service-account@project.iam.gserviceaccount.com",
    PrivateKeyPEM:       rsaPrivateKeyPEMBytes,
    ChunkSize:           8 * 1024 * 1024, // 8 MiB chunks (multiple of 256 KiB)
    PublicBaseURL:       "https://cdn.example.com", // Optional public CDN prefix
})
j. SFTP / SSH Storage Driver (Remote Linux/Unix Server Storage)

Native SFTP / SSH storage driver connecting BlobKit to remote Linux/Unix file servers, private NAS appliances, and secure file transfer clusters. Supports SSH password or private key authentication (with optional passphrase), atomic temporary file staging, byte-range seeks, sidecar metadata persistence, chroot containment defenses, and multiplexed SSH sessions:

import "github.com/suhwr/blobkit/provider/sftp"

sftpDriver, err := sftp.NewDriver(sftp.Config{
    Name:              "sftp-archive",
    Host:              "storage.internal.corp",
    Port:              22,
    User:              "blobkit-service",
    PrivateKeyPEM:     sshPrivateKeyPEMBytes,
    BaseDir:           "/var/data/blobs",
    EnableSidecarMeta: true,                     // Persist .meta.json sidecars for custom metadata
    PublicBaseURL:     "https://cdn.example.com", // Optional public HTTP gateway prefix
})
2. Advanced Multi-Provider Routing

BlobKit separates provider selection into composable router strategies.

a. Namespace Tiered Routing

Route different workloads to specific storage providers (e.g. public media to R2 for zero egress, compliance archives to AWS S3):

import "github.com/suhwr/blobkit/router"

routes := map[string]blobkit.Driver{
    "public/avatars":    r2Driver,
    "public/media":      r2Driver,
    "secure/documents":  s3Driver,
    "audit/compliance":  s3Driver,
}

nsRouter := router.NewNamespaceRouter(routes, r2Driver /* default fallback */)

client, err := blobkit.New(
    blobkit.WithRouter(nsRouter),
)
b. Active-Passive Failover

Automatically switch from primary provider to secondary backup upon consecutive failure thresholds:

failoverRouter := router.NewFailoverRouter(router.FailoverConfig{
    Primary:          r2Driver,
    Secondary:        s3Driver,
    FailureThreshold: 3,                // Switch to secondary after 3 consecutive failures
    Cooldown:         60 * time.Second, // Probe primary after 60s
})
c. 3-State Circuit Breaker Router

Prevent cascading cloud outages with automatic 3-state circuit breaker protection:

cbRouter := router.NewCircuitBreakerRouter(router.CircuitBreakerConfig{
    Primary:          r2Driver,
    Fallback:         s3Driver,
    FailureThreshold: 5,                // Trip to OPEN after 5 consecutive 5xx errors
    SuccessThreshold: 2,                // Require 2 successful canary probes in HALF-OPEN to close
    Cooldown:         30 * time.Second, // Wait 30s before attempting canary probe
})
d. Weighted Traffic Splitting

Distribute traffic proportionally across multiple cloud providers for canary testing or load balancing:

weightedRouter := router.NewWeightedRouter([]router.WeightedTarget{
    {Driver: r2Driver, Weight: 80}, // 80% of traffic
    {Driver: s3Driver, Weight: 20}, // 20% of traffic
})
e. Capability-Aware Dispatch

Filter storage backends based on required feature bitmasks:

// Only route to drivers that support server-side copy and resumable multipart
capRouter := router.NewCapabilityRouter(
    allDrivers,
    blobkit.CapCopy | blobkit.CapMultipartSession,
)
f. Declarative Multi-Account Fleet Loader (router/fleet)

Orchestrate complete multi-provider fleets and routing topologies declaratively from JSON, YAML, config files, or Go structs without manual wiring. Automatically initializes and wires any combination of all 8 storage drivers (s3, r2, azure, gcs, webdav, gdrive, fs, sftp, memory) with namespace, failover, circuit breaker, or weighted routing:

import "github.com/suhwr/blobkit/router/fleet"

fleetConfigJSON := []byte(`{
  "providers": [
    {
      "type": "s3",
      "name": "r2-public",
      "s3": {
        "endpoint": "https://<account>.r2.cloudflarestorage.com",
        "bucket": "public-media",
        "access_key_id": "...",
        "secret_access_key": "..."
      }
    },
    {
      "type": "azure",
      "name": "azure-archive",
      "azure": {
        "account_name": "prodarchive",
        "account_key": "...",
        "container_name": "archives"
      }
    },
    {
      "type": "fs",
      "name": "local-cache",
      "fs": {
        "root_dir": "/var/data/blobs"
      }
    }
  ],
  "routing": {
    "strategy": "namespace",
    "default_provider": "local-cache",
    "namespaces": {
      "public/media": "r2-public",
      "secure/archive": "azure-archive"
    }
  }
}`)

// Load fleet from JSON or Go struct
storageFleet, err := fleet.LoadFromJSON(fleetConfigJSON)
if err != nil {
    log.Fatalf("failed to initialize storage fleet: %v", err)
}
defer storageFleet.Close()

// Create fully wired BlobKit client directly
client, err := storageFleet.NewClient()
3. Object Key Partitioning Strategies

BlobKit shields applications from manually constructing raw storage paths:

import "github.com/suhwr/blobkit/key"

// 1. UUIDv7 Generator (Logical, time-ordered, K-sortable)
uuidGen := key.NewUUIDv7Generator()

// 2. Date-Partitioned Generator: "namespace/2026/10/06/<uuid>.<ext>"
dateGen := key.NewDatePrefixGenerator()

// 3. Hash-Sharded Generator (Prevents S3 partition rate-limit hotspots):
// Generates keys like: "namespace/a1/b2/<uuid>.<ext>"
shardedGen := key.NewHashShardedGenerator(2, 2)

client, _ := blobkit.New(
    blobkit.WithDriver(r2Driver),
    blobkit.WithKeyGenerator(shardedGen),
)
4. Security Policy & Sanitization Engine

Defend against malicious file uploads, extension spoofing, path traversal, and HTTP header injections at the infrastructure boundary:

import "github.com/suhwr/blobkit/policy"

uploadPolicy := policy.NewBuilder().
    // Allowed MIME types
    AllowMIME("image/jpeg", "image/png", "image/webp", "application/pdf").
    // Allowed extensions
    AllowExtensions(".jpg", ".jpeg", ".png", ".webp", ".pdf").
    // Enforce size bounds (1KB to 25MB)
    MinSize(1024).
    MaxSize(25 * 1024 * 1024).
    // Reject files where file extension doesn't match byte magic signature
    DisallowMIMEMismatch(true).
    Build()

client, _ := blobkit.New(
    blobkit.WithDriver(r2Driver),
    blobkit.WithPolicy(uploadPolicy),
)

// Malicious attempts like:
// - Filename: "../../etc/passwd" -> sanitized to "passwd"
// - Filename: "exploit.exe" disguised as "exploit.png" -> rejected with ErrMIMEMismatch
// - File size: 50MB -> rejected with ErrSizeExceeded
5. Streaming Uploads & Integrity Verification

Upload data directly from any io.Reader without buffering the entire payload in memory. BlobKit automatically computes SHA-256 digests on-the-fly and verifies client-provided checksums:

ctx := context.Background()
fileStream, _ := os.Open("large_document.pdf")
defer fileStream.Close()

obj, err := client.Put(ctx, fileStream, blobkit.PutOptions{
    Namespace: "documents",
    OwnerID:   "tenant_456",
    Filename:  "quarterly_report.pdf",
    // Optional: Assert expected SHA-256 digest
    // If the streamed bytes do not match, BlobKit immediately purges the blob
    // and returns ErrChecksumMismatch
    ClientChecksum: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
    Metadata: map[string]string{
        "department": "finance",
        "confidential": "true",
    },
})
if err != nil {
    if blobkit.IsChecksumMismatch(err) {
        log.Fatalf("Checksum mismatch detected: file was corrupted during transit")
    }
    log.Fatal(err)
}

fmt.Printf("Committed ObjectID: %s\n", obj.ID)
fmt.Printf("Storage Key:        %s\n", obj.Key)
fmt.Printf("SHA-256 Checksum:   %s\n", obj.ChecksumSHA256)
fmt.Printf("Detected MIME:      %s\n", obj.MIMEType)
6. Streaming Downloads & Byte-Range Seeking

Read objects efficiently with support for HTTP byte ranges (essential for video seek, audio streaming, and resumable downloads):

// 1. Full stream download by logical ObjectID
reader, err := client.Get(ctx, obj.ID, blobkit.GetOptions{})
if err != nil {
    log.Fatal(err)
}
defer reader.Close()
io.Copy(destFile, reader)

// 2. Byte-Range request (fetch bytes 1024 to 4096)
rangeReader, err := client.Get(ctx, obj.ID, blobkit.GetOptions{
    Range: "bytes=1024-4096",
})
if err != nil {
    log.Fatal(err)
}
defer rangeReader.Close()
7. High-Level Object Operations

Execute advanced object manipulations purely using logical ObjectIDs.

a. Server-Side & Cross-Driver Copy

Fast native copy when source and destination share the same backend, with automated fallback to streaming copy across different providers:

copiedObj, err := client.Copy(ctx, sourceObj.ID, blobkit.PutOptions{
    Namespace: "documents/archive",
    Filename:  "archived_report.pdf",
})
b. Atomic Move

Coordinate copy and cleanup in a single operation:

movedObj, err := client.Move(ctx, sourceObj.ID, blobkit.PutOptions{
    Namespace: "processed",
})
c. Zero-Byte Logical Rename

Update original presentation filenames in the metadata registry with zero byte rewrites on physical storage:

err := client.Rename(ctx, obj.ID, "final_annual_report_v2.pdf")
d. Existence & Head Metadata Inspection

Inspect object metadata without transferring the body payload:

exists, err := client.Exists(ctx, obj.ID)

headObj, err := client.Head(ctx, obj.ID)
fmt.Printf("MIME: %s, Size: %d bytes\n", headObj.MIMEType, headObj.Size)
e. Concurrency-Bounded Batch Deletes

Delete thousands of objects efficiently with bounded concurrency worker pools:

objectIDs := []string{id1, id2, id3, id4, id5}
err := client.DeleteBatch(ctx, objectIDs)

Enforce strict compliance guarantees and prevent accidental data loss:

retentionTime := time.Now().Add(365 * 24 * time.Hour) // 1 year retention

obj, err := client.Put(ctx, dataStream, blobkit.PutOptions{
    Namespace:      "legal-contracts",
    Filename:       "agreement.pdf",
    RetentionUntil: &retentionTime, // Cannot be deleted before this date
    LegalHold:      true,          // Overrides retention; blocks deletion until released
})

// Attempting to delete returns ErrObjectLocked
err = client.Delete(ctx, obj.ID)
if blobkit.IsObjectLocked(err) {
    fmt.Println("Action blocked: object is locked by retention policy or legal hold")
}

// Soft Delete (Moves to trash, hides from Get/Head queries, retains physical bytes)
err = client.SoftDelete(ctx, obj.ID)

// Restore from trash
err = client.Restore(ctx, obj.ID)

// Permanent Purge (Removes physical bytes from storage and deletes metadata record)
err = client.PermanentDelete(ctx, obj.ID)
9. Resumable Multipart Upload Pipeline

Handle massive files (multi-gigabyte videos, backups, datasets) with resumable session state machines:

// 1. Initiate resumable upload session (chunk size: 10MB, expires in 24 hours)
session, err := client.InitiateResumableUpload(ctx, blobkit.PutOptions{
    Namespace: "videos",
    Filename:  "presentation.mp4",
}, 10*1024*1024, 24*time.Hour)

// 2. Upload individual chunks concurrently or sequentially
part1, err := client.UploadPart(ctx, session.ID, 1, chunk1Reader, chunk1Size)
part2, err := client.UploadPart(ctx, session.ID, 2, chunk2Reader, chunk2Size)

// 3. Inspect upload progress
parts, err := client.ListSessionParts(ctx, session.ID)
fmt.Printf("Uploaded %d parts so far\n", len(parts))

// 4. Commit session into finalized object
finalObj, err := client.CommitResumableUpload(ctx, session.ID)
fmt.Printf("Uploaded completed video: %s (%d bytes)\n", finalObj.ID, finalObj.Size)

// Or abort if canceled by user
// err = client.AbortResumableUpload(ctx, session.ID)
10. URL Resolution & Presigned Access

Deliver assets securely without persisting volatile provider URLs into databases:

// 1. Dynamic Public CDN URL Resolution
// Returns "https://cdn.example.com/media/uploads/2026/10/06/..."
publicURL, err := client.ResolveURL(ctx, obj.ID)

// 2. Time-limited Presigned GET URL (e.g. for private downloads)
presignedGet, err := client.PresignGet(ctx, obj.ID, blobkit.PresignOptions{
    Expiry: 15 * time.Minute,
})
fmt.Printf("Temporary Download Link: %s\n", presignedGet.URL)

// 3. Direct-to-Storage Presigned PUT URL (Client-side upload direct from browser/mobile)
presignedPut, err := client.PresignPut(ctx, blobkit.PutOptions{
    Namespace: "avatars",
    Filename:  "user_photo.jpg",
}, blobkit.PresignOptions{
    Expiry: 10 * time.Minute,
})
fmt.Printf("Direct Upload Target: %s\n", presignedPut.URL)
11. Pluggable Metadata Registries (Memory, SQLite3, PostgreSQL)

BlobKit provides first-class, swappable metadata store implementations for blobkit.MetadataStore. Applications can select the storage engine that matches their deployment architecture:

// Option A: In-Memory Store (Zero setup, ephemeral, perfect for tests & workers)
import "github.com/suhwr/blobkit/registry"
memStore := registry.NewMemoryStore()

// Option B: SQLite3 Store (Embedded local ACID database, zero external dependencies)
import "github.com/suhwr/blobkit/registry/sqlite"
sqliteStore, err := sqlite.New(sqlite.Config{
    FilePath:    "metadata.db", // Or ":memory:" for tests
    BusyTimeout: 5 * time.Second,
    WAL:         true,
    AutoMigrate: true, // Automatically creates tables and indexes
})

// Option C: PostgreSQL Store (Production relational database)
import "github.com/suhwr/blobkit/registry/postgres"
pgStore, err := postgres.New(postgres.Config{
    DSN:         "postgres://user:pass@localhost:5432/app_db?sslmode=disable",
    MaxOpenConns: 25,
    AutoMigrate:  true, // Automatically creates tables and indexes
})

// Initialize BlobKit with any chosen store:
client, _ := blobkit.New(
    blobkit.WithDriver(r2Driver),
    blobkit.WithRegistry(sqliteStore), // Or memStore, pgStore
)
12. Bounded LRU Caching & Stampede Defense

Prevent cache stampedes and accelerate repeated metadata lookups:

import "github.com/suhwr/blobkit/cache"

// Create LRU cache holding up to 10,000 objects with 10-minute TTL
lruCache := cache.NewLRUCache(10000)

client, _ := blobkit.New(
    blobkit.WithDriver(r2Driver),
    blobkit.WithCache(lruCache),
)

// Repeated calls to Head, Exists, and ResolveURL resolve in < 500ns from memory.
// Mutating operations (Put, Delete, Move, PermanentDelete) automatically invalidate
// the corresponding cache keys with zero race conditions.
13. Decoupled Background Sweeper

Reclaim storage space and eliminate abandoned upload costs using the decoupled sweeper engine:

import "github.com/suhwr/blobkit/lifecycle"

// Configure background sweeper
sweeper, err := lifecycle.NewSweeper(client, lifecycle.SweeperConfig{
    Interval:          1 * time.Hour,
    SoftDeleteTTL:     30 * 24 * time.Hour, // Permanently purge trash after 30 days
    MultipartStaleTTL: 12 * time.Hour,      // Abort abandoned uploads after 12 hours
    BatchSize:         200,
    DryRun:            false,
})
if err != nil {
    log.Fatal(err)
}

// Option A: Run continuously in background goroutine
_ = sweeper.Start(ctx)
defer sweeper.Stop()

// Option B: Run as a one-off scheduled Kubernetes CronJob
// err := sweeper.RunOnce(ctx)
14. Vendor-Neutral Telemetry & Observability

Monitor throughput, error rates, latencies, and memory semaphore wait times:

import "github.com/suhwr/blobkit/observer"

// Built-in thread-safe metrics collector
collector := observer.NewMetricsCollector()

client, _ := blobkit.New(
    blobkit.WithDriver(r2Driver),
    blobkit.WithObserver(collector),
)

// Inspect metrics anytime
metrics := collector.Snapshot()
fmt.Printf("Total Uploads:     %d\n", metrics.Operations["put"])
fmt.Printf("Bytes Transferred: %d\n", metrics.BytesTransferred)
fmt.Printf("Active Semaphores: %d\n", metrics.LimiterWaits)

To export to Prometheus or OpenTelemetry, simply implement the blobkit.Observer interface:

type Observer interface {
    OnOperationStart(ctx context.Context, op string, namespace string)
    OnOperationEnd(ctx context.Context, op string, namespace string, provider string, duration time.Duration, err error)
    OnBytesTransferred(ctx context.Context, op string, bytes int64)
    OnLimiterWait(ctx context.Context, waitDuration time.Duration)
}
15. Consumer Unit Testing with Test Fixtures

Test consumer applications without external network calls, cloud credentials, or MinIO containers using testutil:

package service_test

import (
    "testing"
    "github.com/suhwr/blobkit/testutil"
)

func TestUserAvatarUpload(t *testing.T) {
    // Spin up an isolated, in-memory BlobKit test fixture
    fixture := testutil.NewMockStorage()
    defer fixture.Client.Close()

    // Seed existing mock object
    seeded := fixture.SeedObject("avatars", "existing.png", []byte("avatar-bytes"))

    // Execute application service under test
    userService := NewUserService(fixture.Client)
    err := userService.UpdateAvatar(seeded.ID)
    if err != nil {
        t.Fatalf("UpdateAvatar failed: %v", err)
    }

    // Assert stored contents
    testutil.AssertBlobContent(t, fixture.Client, seeded.ID, []byte("avatar-bytes"))
}

Performance Benchmarks

Measured on AMD EPYC 7C13 Processor (8 CPU cores allocated, Linux x86_64, Go 1.27):

Benchmark Scenario Throughput / Ops Latency (ns/op) Memory (B/op) Allocs/op Architectural Role
BenchmarkCircuitBreaker_Select-8 86,390,000+ 13.92 ns/op 0 B/op 0 allocs/op Zero-allocation route decision
BenchmarkLRUCache_Hit-8 4,640,000+ 279.4 ns/op 320 B/op 1 allocs/op Sub-microsecond metadata cache
BenchmarkKeyGeneration_UUIDv7-8 2,270,000+ 441.3 ns/op 208 B/op 4 allocs/op Time-ordered collision-proof keys
BenchmarkMIMESniff-8 2,050,000+ 593.7 ns/op 624 B/op 3 allocs/op 512B zero-rewind magic signature check
BenchmarkKeyGeneration_HashSharded-8 1,720,000+ 696.2 ns/op 416 B/op 7 allocs/op High-cardinality directory partitioning
BenchmarkKeyGeneration_DatePrefix-8 1,760,000+ 711.6 ns/op 256 B/op 5 allocs/op YYYY/MM/DD prefix layout
BenchmarkBucket_Get-8 1,530,000+ 735.4 ns/op 450 B/op 5 allocs/op Low-level direct key read (Bucket)
BenchmarkClient_Get_ByObjectID-8 1,000,000+ 1,124 ns/op 739 B/op 5 allocs/op Logical ID resolution + read (Client)
BenchmarkBucket_Put-8 484,000+ 2,335 ns/op 1,537 B/op 12 allocs/op Low-level direct wire upload (3.3x faster)
BenchmarkClient_Put_Standalone-8 163,000+ 7,706 ns/op 3,234 B/op 33 allocs/op MIME sniff + SHA256 TeeReader + KeyGen
BenchmarkClient_Put_WithRegistry-8 120,000+ 9,905 ns/op 4,509 B/op 36 allocs/op Full pipeline + in-memory SQL state machine

License

MIT License. See LICENSE for details.

Documentation ¶

Index ¶

Constants ¶

View Source 
const DefaultPresignExpiry = 15 * time.Minute

DefaultPresignExpiry is the default duration for generated presigned URLs.

View Source 
const MaxKeyLength = 1024

MaxKeyLength defines the maximum permitted byte length for an object key (1024 bytes, conforming to cloud standards).

View Source 
const SizeUnknown int64 = -1

SizeUnknown represents an unknown or unbounded streaming payload size.

Variables ¶

View Source 
var (
	// ErrObjectNotFound indicates that the requested object does not exist in storage.
	ErrObjectNotFound = errors.New("blobkit: object not found")

	// ErrNotFound is an alias for ErrObjectNotFound for ergonomic parity.
	ErrNotFound = ErrObjectNotFound

	// ErrBucketNotFound indicates that the target bucket does not exist.
	ErrBucketNotFound = errors.New("blobkit: bucket not found")

	// ErrInvalidKey indicates that the provided object key is empty or malformed.
	ErrInvalidKey = errors.New("blobkit: invalid object key")

	// ErrInvalidID indicates that the provided object identifier is malformed.
	ErrInvalidID = errors.New("blobkit: invalid object id")

	// ErrPreconditionFailed indicates that an ETag or If-Match condition was not met.
	ErrPreconditionFailed = errors.New("blobkit: precondition failed")

	// ErrProviderUnavailable indicates that the storage backend is unreachable or returning 5xx.
	ErrProviderUnavailable = errors.New("blobkit: provider unavailable")

	// ErrQuotaExceeded indicates that the storage account quota or size limit was reached.
	ErrQuotaExceeded = errors.New("blobkit: storage quota exceeded")

	// ErrUploadTooLarge indicates that the payload exceeds the provider's maximum single or multipart upload limit.
	ErrUploadTooLarge = errors.New("blobkit: upload size exceeds maximum allowed")

	// ErrUnsupportedOperation indicates that the provider does not support the requested operation.
	ErrUnsupportedOperation = errors.New("blobkit: unsupported operation")

	// ErrNotSupported is an alias for ErrUnsupportedOperation.
	ErrNotSupported = ErrUnsupportedOperation

	// ErrMemoryBudgetExceeded indicates that the global or stream multipart memory budget has been exhausted.
	ErrMemoryBudgetExceeded = errors.New("blobkit: memory budget exceeded")

	// ErrNilReader indicates that a nil io.Reader was passed to an upload operation.
	ErrNilReader = errors.New("blobkit: nil reader provided")

	// ErrSecurityViolation indicates that an operation violated security policies (e.g. traversal, forbidden extensions).
	ErrSecurityViolation = errors.New("blobkit: security violation")

	// ErrMIMEMismatch indicates that the content stream magic bytes contradict the claimed file extension.
	ErrMIMEMismatch = errors.New("blobkit: MIME type mismatch with content")

	// ErrInvalidFilename indicates that the provided filename is empty, malformed, or too long.
	ErrInvalidFilename = errors.New("blobkit: invalid filename")

	// ErrChecksumMismatch indicates that the computed payload hash does not match the expected checksum.
	ErrChecksumMismatch = errors.New("blobkit: checksum mismatch")

	// ErrSizeMismatch indicates that the uploaded byte count differs from the declared size.
	ErrSizeMismatch = errors.New("blobkit: size mismatch")

	// ErrObjectLocked indicates that retention policy or active legal hold prohibits modification or deletion.
	ErrObjectLocked = errors.New("blobkit: object retention locked or legal hold active")

	// ErrSessionNotFound indicates that the referenced resumable multipart upload session does not exist.
	ErrSessionNotFound = errors.New("blobkit: upload session not found")

	// ErrSessionExpired indicates that the resumable multipart upload session has expired.
	ErrSessionExpired = errors.New("blobkit: upload session expired")

	// ErrRateLimited indicates that the storage backend rate limit or throttling threshold was exceeded.
	ErrRateLimited = errors.New("blobkit: rate limit exceeded / throttled")

	// ErrPermissionDenied indicates that access to the bucket or object was forbidden.
	ErrPermissionDenied = errors.New("blobkit: permission denied")

	// ErrInvalidCredentials indicates that authentication credentials were missing, expired, or invalid.
	ErrInvalidCredentials = errors.New("blobkit: invalid credentials")

	// ErrMultipartInvalidState indicates an illegal state transition in a multipart upload session.
	ErrMultipartInvalidState = errors.New("blobkit: multipart invalid state transition")
)
View Source 
var (
	// DefaultMaxFilenameLength is the maximum allowed client filename length.
	DefaultMaxFilenameLength = 255
)

Functions ¶

func CheckPreconditions ¶ 

func CheckPreconditions(etag string, modTime time.Time, opts GetOptions) error

CheckPreconditions validates HTTP conditional request headers against object state according to RFC 7232 specifications, returning ErrPreconditionFailed if any condition is unmet.

func EscapeURLPath ¶ 

func EscapeURLPath(key string) string

EscapeURLPath escapes each segment of a key path with url.PathEscape, preserving '/' path separators.

func FormatError ¶ 

func FormatError(err error) string

FormatError returns a sanitized error string.

func IsNotFound ¶ 

func IsNotFound(err error) bool

IsNotFound reports whether err represents an object or bucket not found error.

func IsObjectLocked ¶ 

func IsObjectLocked(err error) bool

IsObjectLocked reports whether err represents a retention lock or legal hold block.

func IsPermanent ¶ 

func IsPermanent(err error) bool

IsPermanent reports whether err represents a definitive, non-recoverable client error that should not be retried and should not trigger failover circuits.

func IsPermissionDenied ¶ 

func IsPermissionDenied(err error) bool

IsPermissionDenied reports whether err represents authentication or authorization failure.

func IsPreconditionFailed ¶ 

func IsPreconditionFailed(err error) bool

IsPreconditionFailed reports whether err represents a precondition failure.

func IsProviderUnavailable ¶ 

func IsProviderUnavailable(err error) bool

IsProviderUnavailable reports whether err represents a backend outage or network failure.

func IsQuotaExceeded ¶ 

func IsQuotaExceeded(err error) bool

IsQuotaExceeded reports whether err represents a storage quota limit error.

func IsRateLimited ¶ 

func IsRateLimited(err error) bool

IsRateLimited reports whether err represents backend rate limiting or throttling.

func IsSecurityViolation ¶ 

func IsSecurityViolation(err error) bool

IsSecurityViolation reports whether err represents a security or validation policy rejection.

func IsTransient ¶ 

func IsTransient(err error) bool

IsTransient reports whether err represents a transient failure eligible for retry or failover. Permanent client errors (such as not found, invalid key, precondition failed, permission denied, checksum mismatch, size mismatch, security violation) are NOT transient.

func NewErrorf ¶ 

func NewErrorf(op, key, provider, format string, a ...any) error

NewErrorf constructs a formatted StorageError.

func PreserveSentinel ¶ 

func PreserveSentinel(err error) bool

PreserveSentinel reports whether err matches a domain sentinel or standard context cancellation error that should be preserved as-is without being wrapped into a generic errors.New().

func ResolvePayload ¶ 

func ResolvePayload(r io.Reader, opts PutOptions) (io.Reader, int64, bool, error)

ResolvePayload inspects the reader and PutOptions to determine exact size semantics. It handles the distinction between known 0-byte objects and unassigned stream sizes.

func SanitizeErrorMessage ¶ 

func SanitizeErrorMessage(raw string) string

SanitizeErrorMessage strips internal proxy dials, local IPs, authorization headers, private keys, and secret tokens from error strings so they are safe for client logging and user-facing error reporting.

func SanitizeFilename ¶ 

func SanitizeFilename(raw string) string

SanitizeFilename removes directory components, null bytes, CRLF characters, and path traversals.

func SanitizeHeader ¶ 

func SanitizeHeader(raw string) string

SanitizeHeader removes CRLF characters and null bytes to prevent HTTP response splitting / header injection.

func ValidateKey ¶ 

func ValidateKey(key string) error

ValidateKey validates an object key against directory traversal, path escape, injection attacks, double/multi-encoding tricks, control characters, and malformed formatting across all operations and storage engines.

func WrapError ¶ 

func WrapError(op, key, provider string, err error) error

WrapError encapsulates an error with operation context and sanitization.

Types ¶

type BaseDriver ¶ 

type BaseDriver struct {
	// contains filtered or unexported fields
}

BaseDriver provides a normative, contract-compliant base implementation of Driver. Embed BaseDriver in your custom storage drivers to automatically fulfill all optional capabilities with standardized ErrUnsupportedOperation errors according to the Driver contract.

By embedding BaseDriver, authors of custom storage drivers only need to implement the operations their backend actually supports (e.g. Put, Get, Delete), while guaranteeing that unadvertised capabilities strictly conform to contract requirements.

func NewBaseDriver ¶ 

func NewBaseDriver(name string, caps Capability) BaseDriver

NewBaseDriver initializes a BaseDriver with a deterministic driver identifier and advertised capabilities.

func (*BaseDriver) AbortMultipart ¶ 

func (b *BaseDriver) AbortMultipart(ctx context.Context, key string, uploadID string) error

AbortMultipart is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) Capabilities ¶ 

func (b *BaseDriver) Capabilities() Capability

Capabilities returns the bitmask of features supported by the driver.

func (*BaseDriver) Close ¶ 

func (b *BaseDriver) Close() error

Close gracefully closes the driver. Default is a no-op returning nil.

func (*BaseDriver) CompleteMultipart ¶ 

func (b *BaseDriver) CompleteMultipart(ctx context.Context, obj *Object, uploadID string, parts []CompletedPart) (*Object, error)

CompleteMultipart is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) Copy ¶ 

func (b *BaseDriver) Copy(ctx context.Context, srcKey, dstKey string) error

Copy is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) CreateMultipart ¶ 

func (b *BaseDriver) CreateMultipart(ctx context.Context, obj *Object, opts PutOptions) (string, error)

CreateMultipart is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) Delete ¶ 

func (b *BaseDriver) Delete(ctx context.Context, key string) error

Delete is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) DeleteBatch ¶ 

func (b *BaseDriver) DeleteBatch(ctx context.Context, keys []string) ([]string, error)

DeleteBatch is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) Get ¶ 

func (b *BaseDriver) Get(ctx context.Context, key string, opts GetOptions) (*ObjectReader, error)

Get is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) Head ¶ 

func (b *BaseDriver) Head(ctx context.Context, key string) (*Object, error)

Head is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) List ¶ 

func (b *BaseDriver) List(ctx context.Context, opts ListOptions) (*ListResult, error)

List is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) ListParts ¶ 

func (b *BaseDriver) ListParts(ctx context.Context, key string, uploadID string) ([]CompletedPart, error)

ListParts is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) Name ¶ 

func (b *BaseDriver) Name() string

Name returns the driver identifier.

func (*BaseDriver) PresignGet ¶ 

func (b *BaseDriver) PresignGet(ctx context.Context, key string, opts PresignOptions) (*PresignedURL, error)

PresignGet is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) PresignPut ¶ 

func (b *BaseDriver) PresignPut(ctx context.Context, key string, opts PresignOptions) (*PresignedURL, error)

PresignPut is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) Put ¶ 

func (b *BaseDriver) Put(ctx context.Context, obj *Object, r io.Reader, opts PutOptions) (*Object, error)

Put is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) ResolveURL ¶ 

func (b *BaseDriver) ResolveURL(key string) (string, error)

ResolveURL is a default stub that returns ErrUnsupportedOperation.

func (*BaseDriver) UploadPart ¶ 

func (b *BaseDriver) UploadPart(ctx context.Context, key string, uploadID string, partNumber int32, r io.Reader, size int64) (string, error)

UploadPart is a default stub that returns ErrUnsupportedOperation.

type Bucket ¶ 

type Bucket struct {
	// contains filtered or unexported fields
}

Bucket provides a lightweight, stateless, key-value object storage abstraction. It bypasses the database metadata registry, canonical ObjectID translations, and background lifecycle management of Client, communicating directly with a Driver using raw storage keys (e.g. "avatars/user-123.png").

Use Bucket when your application needs high-throughput, direct wire-level object storage without database dependencies or opinionated key generation.

func NewBucket ¶ 

func NewBucket(driver Driver) (*Bucket, error)

NewBucket constructs a low-level Bucket backed by the provided Driver.

func (*Bucket) AbortMultipart ¶ 

func (b *Bucket) AbortMultipart(ctx context.Context, key string, uploadID string) error

AbortMultipart cancels a multi-part upload session and cleans up staged parts.

func (*Bucket) Capabilities ¶ 

func (b *Bucket) Capabilities() Capability

Capabilities returns the supported capability bitmask of the driver.

func (*Bucket) Close ¶ 

func (b *Bucket) Close() error

Close gracefully closes the underlying Driver.

func (*Bucket) CompleteMultipart ¶ 

func (b *Bucket) CompleteMultipart(ctx context.Context, key string, uploadID string, parts []CompletedPart) (*Object, error)

CompleteMultipart finalizes an ongoing multi-part upload session.

func (*Bucket) Copy ¶ 

func (b *Bucket) Copy(ctx context.Context, srcKey, dstKey string) error

Copy duplicates an object within this bucket.

func (*Bucket) CreateMultipart ¶ 

func (b *Bucket) CreateMultipart(ctx context.Context, key string, opts PutOptions) (uploadID string, err error)

CreateMultipart initializes a multi-part upload session directly on the storage provider.

func (*Bucket) Delete ¶ 

func (b *Bucket) Delete(ctx context.Context, key string) error

Delete permanently removes an object by key.

func (*Bucket) DeleteBatch ¶ 

func (b *Bucket) DeleteBatch(ctx context.Context, keys []string) ([]string, error)

DeleteBatch removes multiple keys in a single operation.

func (*Bucket) Driver ¶ 

func (b *Bucket) Driver() Driver

Driver returns the underlying Driver instance.

func (*Bucket) Exists ¶ 

func (b *Bucket) Exists(ctx context.Context, key string) (bool, error)

Exists reports whether an object currently exists at the specified key.

func (*Bucket) Get ¶ 

func (b *Bucket) Get(ctx context.Context, key string, opts GetOptions) (*ObjectReader, error)

Get retrieves an object stream and its metadata.

func (*Bucket) GetBytes ¶ 

func (b *Bucket) GetBytes(ctx context.Context, key string, opts GetOptions) ([]byte, *Object, error)

GetBytes reads the entire object payload into an in-memory byte slice.

func (*Bucket) Head ¶ 

func (b *Bucket) Head(ctx context.Context, key string) (*Object, error)

Head retrieves metadata for an object without reading its payload body.

func (*Bucket) List ¶ 

func (b *Bucket) List(ctx context.Context, opts ListOptions) (*ListResult, error)

List queries objects matching prefix and filter criteria.

func (*Bucket) ListParts ¶ 

func (b *Bucket) ListParts(ctx context.Context, key string, uploadID string) ([]CompletedPart, error)

ListParts returns the list of parts already committed to an active multipart session.

func (*Bucket) Name ¶ 

func (b *Bucket) Name() string

Name returns the driver identifier (e.g. "s3-primary", "r2-edge", "memory").

func (*Bucket) NewWriter ¶ 

func (b *Bucket) NewWriter(ctx context.Context, key string, opts PutOptions) (*BucketWriter, error)

NewWriter creates an io.WriteCloser that streams data directly to the given key. The upload executes concurrently as bytes are written. Calling Close() finalizes the upload.

func (*Bucket) OpenSeeker ¶ 

func (b *Bucket) OpenSeeker(ctx context.Context, key string) (*SeekableReader, error)

OpenSeeker creates an io.ReadSeekCloser and io.ReaderAt over the given object key. It verifies the driver's CapByteRangeGet capability and fetches the object's total size via Head.

func (*Bucket) PresignGet ¶ 

func (b *Bucket) PresignGet(ctx context.Context, key string, opts PresignOptions) (*PresignedURL, error)

PresignGet generates a temporary signed download URL for direct client access.

func (*Bucket) PresignPut ¶ 

func (b *Bucket) PresignPut(ctx context.Context, key string, opts PresignOptions) (*PresignedURL, error)

PresignPut generates a temporary signed upload URL for direct browser-to-storage uploads.

func (*Bucket) Put ¶ 

func (b *Bucket) Put(ctx context.Context, key string, r io.Reader, opts PutOptions) (*Object, error)

Put uploads an object stream directly to the specified key.

func (*Bucket) PutBytes ¶ 

func (b *Bucket) PutBytes(ctx context.Context, key string, data []byte, opts PutOptions) (*Object, error)

PutBytes is a convenience method to store an in-memory byte slice directly.

func (*Bucket) ResolveURL ¶ 

func (b *Bucket) ResolveURL(key string) (string, error)

ResolveURL builds a public access or CDN delivery URL for the given key.

func (*Bucket) UploadPart ¶ 

func (b *Bucket) UploadPart(ctx context.Context, key string, uploadID string, partNumber int32, r io.Reader, size int64) (etag string, err error)

UploadPart uploads a single part of an ongoing multi-part upload session.

type BucketWriter ¶ 

type BucketWriter struct {
	// contains filtered or unexported fields
}

BucketWriter implements io.WriteCloser for streaming writes directly into storage.

func (*BucketWriter) Close ¶ 

func (w *BucketWriter) Close() error

Close finalizes the upload stream and waits for the storage backend to confirm persistence.

func (*BucketWriter) CloseWithError ¶ 

func (w *BucketWriter) CloseWithError(err error) error

CloseWithError aborts the streaming upload with the specified error.

func (*BucketWriter) Write ¶ 

func (w *BucketWriter) Write(p []byte) (n int, err error)

Write writes a slice of bytes to the underlying streaming pipeline.

type CDNResolver ¶ 

type CDNResolver struct {
	// contains filtered or unexported fields
}

CDNResolver resolves object keys against a base CDN or public domain URL (e.g. "https://cdn.example.com").

func NewCDNResolver ¶ 

func NewCDNResolver(baseURL string) *CDNResolver

NewCDNResolver initializes a CDNResolver. Trailing slashes are stripped from baseURL.

func (*CDNResolver) ResolveURL ¶ 

func (c *CDNResolver) ResolveURL(key string) string

ResolveURL appends the sanitized key to the configured base URL.

type Cache ¶ 

type Cache interface {
	// Get retrieves cached metadata for the given storage key or object ID.
	Get(key string) (*Object, bool)

	// Set caches object metadata with a specific TTL.
	Set(key string, obj *Object, ttl time.Duration)

	// SetNegative records a negative cache entry (not found) to shield storage backends.
	SetNegative(key string, ttl time.Duration)

	// IsNegative checks whether a key is currently negative-cached.
	IsNegative(key string) bool

	// Delete invalidates any cached entries for the specified key.
	Delete(key string)

	// Purge clears all entries in the cache.
	Purge()
}

Cache defines an optional cache interface for storing object metadata and negative lookups.

type Capability ¶ 

type Capability uint32

Capability is a bitmask representing features supported by a storage driver. 

const (
	// CapDirectPut indicates support for single-shot uploads.
	CapDirectPut Capability = 1 << iota

	// CapMultipartPut indicates support for internal chunked multipart uploads.
	CapMultipartPut

	// CapPresignGet indicates support for generating presigned GET URLs.
	CapPresignGet

	// CapPresignPut indicates support for generating presigned PUT URLs.
	CapPresignPut

	// CapBatchDelete indicates support for deleting multiple keys in a single operation.
	CapBatchDelete

	// CapByteRangeGet indicates support for HTTP Range requests.
	CapByteRangeGet

	// CapCopy indicates support for server-side object duplication.
	CapCopy

	// CapMultipartSession indicates support for resumable multi-part upload primitives.
	CapMultipartSession
)

type Client ¶ 

type Client struct {
	// contains filtered or unexported fields
}

Client is the high-level object storage infrastructure orchestrator. It sits between application logic and physical storage drivers, managing identity, key generation, streaming I/O, routing, metadata, and access delivery.

func New ¶ 

func New(options ...Option) (*Client, error)

New constructs a new BlobKit Client configured with the provided options.

func (*Client) AbortResumableUpload ¶ 

func (c *Client) AbortResumableUpload(ctx context.Context, sessionID string) error

AbortResumableUpload aborts an in-progress multipart session and purges partial chunks.

func (*Client) BatchUpdateMetadata ¶ 

func (c *Client) BatchUpdateMetadata(ctx context.Context, updates map[string]map[string]string) error

BatchUpdateMetadata applies metadata key-value updates across multiple objects in the registry.

func (*Client) Bucket ¶ 

func (c *Client) Bucket(ctx context.Context, providerName string) (*Bucket, error)

Bucket returns a low-level Bucket bound to the specified provider name, or the default driver resolved by the router if providerName is empty.

func (*Client) Close ¶ 

func (c *Client) Close() error

Close gracefully releases any connections across all registered drivers and the registry.

func (*Client) Commit ¶ 

func (c *Client) Commit(ctx context.Context, objectID string) error

Commit verifies that an uploaded blob exists in storage and transitions its status from StatePending to StateCommitted in the database registry.

func (*Client) CommitResumableUpload ¶ 

func (c *Client) CommitResumableUpload(ctx context.Context, sessionID string) (*Object, error)

CommitResumableUpload finalizes an active resumable upload session, assembling all parts into a completed object in storage and transitioning its registry record to StateCommitted.

func (*Client) Copy ¶ 

func (c *Client) Copy(ctx context.Context, srcTarget string, dstOpts PutOptions) (*Object, error)

Copy duplicates an object to a new logical destination. If the source and destination drivers are identical and support CapCopy, a fast server-side copy is executed; otherwise, data is streamed between drivers.

func (*Client) Delete ¶ 

func (c *Client) Delete(ctx context.Context, target string) (err error)

Delete removes an object from storage and marks or deletes it in the registry.

func (*Client) DeleteBatch ¶ 

func (c *Client) DeleteBatch(ctx context.Context, targets []string) (deleted []string, err error)

DeleteBatch removes multiple objects in a single batch operation across their respective storage providers.

func (*Client) Exists ¶ 

func (c *Client) Exists(ctx context.Context, target string) (bool, error)

Exists reports whether an object exists in storage without downloading its body.

func (*Client) Find ¶ 

func (c *Client) Find(ctx context.Context, filter Filter) ([]Object, error)

Find queries objects by metadata attributes (namespace, owner_id, status, mime, etc.) from the database registry without requiring knowledge of storage keys or URLs.

func (*Client) FindExpiredObjects ¶ 

func (c *Client) FindExpiredObjects(ctx context.Context, before time.Time, limit int) ([]Record, error)

FindExpiredObjects returns records whose scheduled expiration has elapsed.

func (*Client) FindSoftDeletedObjects ¶ 

func (c *Client) FindSoftDeletedObjects(ctx context.Context, before time.Time, limit int) ([]Record, error)

FindSoftDeletedObjects returns records marked as soft-deleted before the specified cutoff.

func (*Client) FindStaleSessions ¶ 

func (c *Client) FindStaleSessions(ctx context.Context, before time.Time, limit int) ([]UploadSession, error)

FindStaleSessions returns active sessions that have passed their expiration timestamp.

func (*Client) Get ¶ 

func (c *Client) Get(ctx context.Context, target string, opts GetOptions) (reader *ObjectReader, err error)

Get retrieves an object stream. Target can be either a canonical logical ObjectID (when registry is enabled) or a physical storage key.

func (*Client) Head ¶ 

func (c *Client) Head(ctx context.Context, target string) (obj *Object, err error)

Head inspects an object's metadata without reading its payload.

func (*Client) InitiateResumableUpload ¶ 

func (c *Client) InitiateResumableUpload(ctx context.Context, opts PutOptions, partSize int64, ttl time.Duration) (*UploadSession, error)

InitiateResumableUpload initializes a multipart upload session for large files or unstable networks. It assigns an ObjectID, physical storage key, creates a multipart session on the storage driver, and returns an UploadSession handle.

func (*Client) List ¶ 

func (c *Client) List(ctx context.Context, opts ListOptions) (*ListResult, error)

List lists objects directly from the storage driver.

func (*Client) ListSessionParts ¶ 

func (c *Client) ListSessionParts(ctx context.Context, sessionID string) ([]CompletedPart, error)

ListSessionParts retrieves all completed parts for the given upload session.

func (*Client) Move ¶ 

func (c *Client) Move(ctx context.Context, srcTarget string, dstOpts PutOptions) (*Object, error)

Move is a best-effort copy and delete. It is not an atomic rename operation. If the server-side copy succeeds but the deletion fails, the object will exist at both locations.

func (*Client) OpenSeeker ¶ 

func (c *Client) OpenSeeker(ctx context.Context, target string) (*SeekableReader, error)

OpenSeeker returns an io.ReadSeekCloser and io.ReaderAt over the specified target (ObjectID or Key).

func (*Client) PermanentDelete ¶ 

func (c *Client) PermanentDelete(ctx context.Context, target string) error

PermanentDelete physically destroys an object from the storage backend and purges its metadata.

func (*Client) PermanentDeleteBatch ¶ 

func (c *Client) PermanentDeleteBatch(ctx context.Context, targets []string) ([]string, error)

PermanentDeleteBatch removes multiple objects in bulk by grouping them by provider and invoking Driver.DeleteBatch, with automatic metadata cleanup and cache eviction.

func (*Client) PresignGet ¶ 

func (c *Client) PresignGet(ctx context.Context, target string, opts PresignOptions) (*PresignedURL, error)

PresignGet generates a time-limited signed URL for authorized download. Target can be an ObjectID or a physical storage key.

func (*Client) PresignPut ¶ 

func (c *Client) PresignPut(ctx context.Context, opts PutOptions, presignOpts PresignOptions) (*PresignedURL, *Object, error)

PresignPut generates a time-limited signed URL for direct client upload. It assigns an ObjectID, calculates the physical storage key, and registers a pending record.

func (*Client) Put ¶ 

func (c *Client) Put(ctx context.Context, r io.Reader, opts PutOptions) (savedObj *Object, err error)

Put uploads an object stream based on the caller's semantic intent. It assigns a canonical logical ObjectID, generates a physical storage key, peeks MIME types without buffering entire files, and commits metadata to the registry if enabled.

func (*Client) Reconcile ¶ 

func (c *Client) Reconcile(ctx context.Context, dryRun bool) (*ReconciliationReport, error)

Reconcile audits and synchronizes database registry state with physical storage backends. It detects orphan files (present in driver but absent from registry) and ghost records (marked committed in registry but missing in physical storage). If dryRun is false, ghost records are marked as StateDeleted.

func (*Client) Rename ¶ 

func (c *Client) Rename(ctx context.Context, target string, newFilename string) error

Rename updates the logical client-facing filename without moving or re-uploading physical bytes.

func (*Client) ResolveURL ¶ 

func (c *Client) ResolveURL(ctx context.Context, target string) (string, error)

ResolveURL generates a public CDN or base delivery URL on demand. Target can be an ObjectID or a physical storage key.

func (*Client) Restore ¶ 

func (c *Client) Restore(ctx context.Context, target string) error

Restore recovers a soft-deleted object, returning its lifecycle state to StateCommitted.

func (*Client) SoftDelete ¶ 

func (c *Client) SoftDelete(ctx context.Context, target string) error

SoftDelete marks an object as deleted in the registry, hiding it from standard Get/Head queries without immediately destroying the underlying physical storage blob.

func (*Client) UpdateMetadata ¶ 

func (c *Client) UpdateMetadata(ctx context.Context, target string, metadata map[string]string) error

UpdateMetadata replaces or updates user-defined metadata attributes for an object.

func (*Client) UploadPart ¶ 

func (c *Client) UploadPart(ctx context.Context, sessionID string, partNumber int, r io.Reader, size int64) (*CompletedPart, error)

UploadPart uploads a single chunk within an active resumable upload session.

type CompletedPart ¶ 

type CompletedPart struct {
	// PartNumber is the 1-based index of the part.
	PartNumber int32 `json:"part_number"`

	// ETag is the entity tag assigned by the storage provider for this part.
	ETag string `json:"etag"`

	// Size is the part size in bytes.
	Size int64 `json:"size"`
}

CompletedPart records information about an uploaded multipart part.

type CustomResolver ¶ 

type CustomResolver func(key string) string

CustomResolver adapts an arbitrary formatting function into a URLResolver.

func (CustomResolver) ResolveURL ¶ 

func (f CustomResolver) ResolveURL(key string) string

ResolveURL invokes the underlying custom function.

type DelegateDriver ¶ 

type DelegateDriver struct {
	Driver Driver
}

DelegateDriver is a transparent proxy struct that implements Driver by forwarding all calls to its inner Driver. Custom middleware authors should embed DelegateDriver and selectively override only the specific methods they wish to intercept.

func NewDelegateDriver ¶ 

func NewDelegateDriver(target Driver) DelegateDriver

NewDelegateDriver creates a DelegateDriver wrapping the specified target Driver.

func (*DelegateDriver) AbortMultipart ¶ 

func (d *DelegateDriver) AbortMultipart(ctx context.Context, key string, uploadID string) error

AbortMultipart delegates to the inner driver.

func (*DelegateDriver) Capabilities ¶ 

func (d *DelegateDriver) Capabilities() Capability

Capabilities delegates to the inner driver.

func (*DelegateDriver) Close ¶ 

func (d *DelegateDriver) Close() error

Close delegates to the inner driver.

func (*DelegateDriver) CompleteMultipart ¶ 

func (d *DelegateDriver) CompleteMultipart(ctx context.Context, obj *Object, uploadID string, parts []CompletedPart) (*Object, error)

CompleteMultipart delegates to the inner driver.

func (*DelegateDriver) Copy ¶ 

func (d *DelegateDriver) Copy(ctx context.Context, srcKey, dstKey string) error

Copy delegates to the inner driver.

func (*DelegateDriver) CreateMultipart ¶ 

func (d *DelegateDriver) CreateMultipart(ctx context.Context, obj *Object, opts PutOptions) (string, error)

CreateMultipart delegates to the inner driver.

func (*DelegateDriver) Delete ¶ 

func (d *DelegateDriver) Delete(ctx context.Context, key string) error

Delete delegates to the inner driver.

func (*DelegateDriver) DeleteBatch ¶ 

func (d *DelegateDriver) DeleteBatch(ctx context.Context, keys []string) ([]string, error)

DeleteBatch delegates to the inner driver.

func (*DelegateDriver) Get ¶ 

func (d *DelegateDriver) Get(ctx context.Context, key string, opts GetOptions) (*ObjectReader, error)

Get delegates to the inner driver.

func (*DelegateDriver) Head ¶ 

func (d *DelegateDriver) Head(ctx context.Context, key string) (*Object, error)

Head delegates to the inner driver.

func (*DelegateDriver) List ¶ 

func (d *DelegateDriver) List(ctx context.Context, opts ListOptions) (*ListResult, error)

List delegates to the inner driver.

func (*DelegateDriver) ListParts ¶ 

func (d *DelegateDriver) ListParts(ctx context.Context, key string, uploadID string) ([]CompletedPart, error)

ListParts delegates to the inner driver.

func (*DelegateDriver) Name ¶ 

func (d *DelegateDriver) Name() string

Name delegates to the inner driver.

func (*DelegateDriver) PresignGet ¶ 

func (d *DelegateDriver) PresignGet(ctx context.Context, key string, opts PresignOptions) (*PresignedURL, error)

PresignGet delegates to the inner driver.

func (*DelegateDriver) PresignPut ¶ 

func (d *DelegateDriver) PresignPut(ctx context.Context, key string, opts PresignOptions) (*PresignedURL, error)

PresignPut delegates to the inner driver.

func (*DelegateDriver) Put ¶ 

func (d *DelegateDriver) Put(ctx context.Context, obj *Object, r io.Reader, opts PutOptions) (*Object, error)

Put delegates to the inner driver.

func (*DelegateDriver) ResolveURL ¶ 

func (d *DelegateDriver) ResolveURL(key string) (string, error)

ResolveURL delegates to the inner driver.

func (*DelegateDriver) UploadPart ¶ 

func (d *DelegateDriver) UploadPart(ctx context.Context, key string, uploadID string, partNumber int32, r io.Reader, size int64) (string, error)

UploadPart delegates to the inner driver.

type Driver ¶ 

type Driver interface {
	// Name returns the driver identifier (e.g. "r2-primary", "s3-backup", "memory").
	// MUST return a non-empty, deterministic string identifying this driver instance.
	Name() string

	// Capabilities returns the bitmask of operations supported by this driver.
	// MUST accurately declare supported features. A driver MUST NOT declare a capability
	// it cannot fulfill according to the normative contract.
	Capabilities() Capability

	// Put uploads an object stream to the driver backend.
	//
	// Semantic Guarantees:
	// - Inputs: obj MUST be non-nil with a valid Key. r MUST be non-nil.
	// - Missing/Invalid inputs: If r is nil, MUST return ErrNilReader. If obj is nil or Key is invalid, MUST return ErrInvalidKey or ErrSecurityViolation.
	// - Size semantics: If opts.ExplicitSize is true or opts.Size >= 0, the stream MUST yield exactly opts.Size bytes.
	//   If fewer bytes are read, MUST fail with ErrSizeMismatch.
	//   If surplus bytes remain after reading opts.Size bytes, MUST fail with ErrSizeMismatch.
	//   On ErrSizeMismatch or failure, the driver MUST NOT leave partially-created or corrupted objects in storage.
	// - Empty objects: MUST support storing, overwriting, and retrieving valid 0-byte objects.
	// - Overwrites: If an object already exists at obj.Key, Put MUST overwrite it.
	//   Filesystem implementations MUST perform atomic replacement (e.g. via temporary file rename).
	// - Context & Cancellation: If ctx is canceled or times out before or during upload, MUST abort and return ctx.Err().
	//   MUST clean up partial objects on cancellation.
	// - Concurrency: Put MUST be safe for concurrent execution across identical or distinct keys.
	Put(ctx context.Context, obj *Object, r io.Reader, opts PutOptions) (*Object, error)

	// Get retrieves an object stream and its metadata.
	//
	// Semantic Guarantees:
	// - Inputs: key MUST be valid. If key is invalid, MUST return ErrInvalidKey or ErrSecurityViolation.
	// - Missing object: If the object does not exist, MUST return ErrObjectNotFound.
	// - Empty object: If the object is 0 bytes, MUST return an ObjectReader whose Body yields 0 bytes and io.EOF.
	// - Preconditions: If opts contains IfMatch, IfNoneMatch, IfModifiedSince, or IfUnmodifiedSince,
	//   the driver MUST evaluate them. If preconditions fail, MUST return ErrPreconditionFailed.
	// - Byte-Range: If opts.Range is set and CapByteRangeGet is supported, MUST return the requested subslice.
	// - Context & Cancellation: If ctx is canceled, reading or requesting MUST return ctx.Err().
	Get(ctx context.Context, key string, opts GetOptions) (*ObjectReader, error)

	// Head inspects an object and returns its metadata without fetching the body.
	//
	// Semantic Guarantees:
	// - Inputs: key MUST be valid. If invalid, MUST return ErrInvalidKey or ErrSecurityViolation.
	// - Missing object: If the object does not exist, MUST return ErrObjectNotFound.
	// - Empty object: If the object is 0 bytes, MUST return Object with Size == 0 and nil error.
	// - Content: MUST NOT read or consume the body payload.
	Head(ctx context.Context, key string) (*Object, error)

	// Delete removes an object from storage.
	//
	// Semantic Guarantees:
	// - Inputs: key MUST be valid. If invalid, MUST return ErrInvalidKey or ErrSecurityViolation.
	// - Idempotency: Deleting an already-deleted or non-existent key MUST succeed and return nil error.
	// - Overwrite & Concurrency: MUST be safe for concurrent execution.
	// - Context: If ctx is canceled before execution, MUST return ctx.Err().
	Delete(ctx context.Context, key string) error

	// DeleteBatch removes multiple keys in a batch, returning deleted keys.
	//
	// Semantic Guarantees:
	// - CapBatchDelete: Only invoked if driver declares CapBatchDelete.
	// - Empty input: If keys is empty, MUST return empty list and nil error.
	// - Partial failure: MUST return the list of successfully deleted keys alongside any error encountered.
	DeleteBatch(ctx context.Context, keys []string) ([]string, error)

	// Copy duplicates an object within this driver backend.
	//
	// Semantic Guarantees:
	// - CapCopy: Only invoked if driver declares CapCopy.
	// - Missing source: If srcKey does not exist, MUST return ErrObjectNotFound.
	// - Destination: Overwrites dstKey if it exists. Source object MUST remain unchanged.
	Copy(ctx context.Context, srcKey, dstKey string) error

	// List queries objects matching criteria.
	//
	// Semantic Guarantees:
	// - Prefix & Delimiter: MUST accurately filter by prefix and group common prefixes when delimiter is provided.
	List(ctx context.Context, opts ListOptions) (*ListResult, error)

	// PresignGet creates a temporary signed URL for object download.
	// CapPresignGet: Only invoked if driver declares CapPresignGet.
	PresignGet(ctx context.Context, key string, opts PresignOptions) (*PresignedURL, error)

	// PresignPut creates a temporary signed URL for direct object upload.
	// CapPresignPut: Only invoked if driver declares CapPresignPut.
	PresignPut(ctx context.Context, key string, opts PresignOptions) (*PresignedURL, error)

	// ResolveURL builds a public access or CDN URL for the given key.
	ResolveURL(key string) (string, error)

	// CreateMultipart initiates a resumable multi-part upload session on the backend.
	//
	// Semantic Guarantees:
	// - CapMultipartSession: Only invoked if driver declares CapMultipartSession.
	// - Inputs: obj MUST be non-nil with valid Key.
	// - State: Initializes session in StateCreated. Returns unique uploadID.
	CreateMultipart(ctx context.Context, obj *Object, opts PutOptions) (uploadID string, err error)

	// UploadPart uploads a single chunk of a multipart upload.
	//
	// Semantic Guarantees:
	// - Inputs: partNumber MUST be >= 1. size MUST match the exact payload read from r.
	// - Size verification: If r yields fewer or more bytes than size, MUST fail with ErrSizeMismatch.
	// - Session existence: If uploadID does not exist or has been completed/aborted, MUST return ErrSessionNotFound.
	UploadPart(ctx context.Context, key string, uploadID string, partNumber int32, r io.Reader, size int64) (etag string, err error)

	// CompleteMultipart finalizes a multipart upload session.
	//
	// Semantic Guarantees:
	// - Inputs: parts MUST contain at least one part (unless backend allows 0-part empty).
	// - Parts validation: parts MUST be ordered or sortable by partNumber without missing or duplicate parts.
	// - State: Transitions session to StateCompleted. Subsequent operations on uploadID MUST fail with ErrSessionNotFound.
	// - Cleanup: MUST clean up temporary staged chunks.
	CompleteMultipart(ctx context.Context, obj *Object, uploadID string, parts []CompletedPart) (*Object, error)

	// AbortMultipart cancels a multipart upload and cleans up stored parts.
	//
	// Semantic Guarantees:
	// - State: Transitions session to StateAborted. Cleans up all staged chunks.
	// - Idempotency: Calling AbortMultipart on an already-aborted session SHOULD return nil error or ErrSessionNotFound.
	AbortMultipart(ctx context.Context, key string, uploadID string) error

	// ListParts returns the list of parts already uploaded for an active session.
	//
	// Semantic Guarantees:
	// - If uploadID does not exist or is aborted/completed, MUST return ErrSessionNotFound.
	ListParts(ctx context.Context, key string, uploadID string) ([]CompletedPart, error)

	// Close gracefully terminates open connections and worker pools.
	Close() error
}

Driver is the storage abstraction interface implemented by backend drivers (e.g. AWS S3, Cloudflare R2, MinIO, Wasabi, In-Memory, POSIX FS, SFTP, WebDAV, GCS, Azure, Google Drive).

Driver Contract (Normative Specification): All Driver implementations MUST conform strictly to the semantic guarantees defined here. Implementations MUST NOT redefine or weaken these guarantees based on backend quirks.

func WrapDriver ¶ 

func WrapDriver(driver Driver, middlewares ...DriverMiddleware) Driver

WrapDriver decorates a target Driver with one or more DriverMiddlewares. Middlewares are executed in the order they are passed: the first middleware in the list is the outermost wrapper, executing first before passing control inward.

type DriverMiddleware ¶ 

type DriverMiddleware func(Driver) Driver

DriverMiddleware wraps an existing Driver with interceptor logic. It allows modifying requests, injecting behaviors (such as encryption, compression, rate limiting, or metrics), or intercepting responses at the wire level.

func LatencyMiddleware ¶ 

func LatencyMiddleware(reporter LatencyReporter) DriverMiddleware

LatencyMiddleware creates a DriverMiddleware that measures and reports the wire-level execution duration of storage driver operations.

type Filter ¶ 

type Filter struct {
	// ObjectID filters by canonical identifier.
	ObjectID string

	// Key filters by physical storage key.
	Key string

	// Namespace filters by category or sub-category (e.g. "bots/autorespon").
	Namespace string

	// OwnerID filters by owner, user, or tenant.
	OwnerID string

	// MIMEType filters by MIME type (e.g. "image/png").
	MIMEType string

	// OriginalFilename filters by user-facing filename.
	OriginalFilename string

	// Status filters by lifecycle state (pending, committed, deleted).
	Status LifecycleState

	// Provider filters by storage driver name.
	Provider string

	// CreatedAfter filters records created on or after this timestamp.
	CreatedAfter *time.Time

	// CreatedBefore filters records created on or before this timestamp.
	CreatedBefore *time.Time

	// DeletedBefore filters records deleted on or before this timestamp.
	DeletedBefore *time.Time

	// Metadata requires matching key-value pairs.
	Metadata map[string]string

	// Limit bounds the result count (default 100, max 1000).
	Limit int

	// Offset enables pagination skipping.
	Offset int
}

Filter specifies query criteria for discovering objects by metadata without needing URLs.

type GetOptions ¶ 

type GetOptions struct {
	// Range specifies an HTTP byte range (e.g. "bytes=0-1024").
	Range string

	// IfMatch returns the object only if its ETag matches this value.
	IfMatch string

	// IfNoneMatch returns the object only if its ETag does not match this value.
	IfNoneMatch string

	// IfModifiedSince returns the object only if it has been modified since this timestamp.
	IfModifiedSince *time.Time

	// IfUnmodifiedSince returns the object only if it has not been modified since this timestamp.
	IfUnmodifiedSince *time.Time
}

GetOptions configures an object download or stream retrieval.

type LatencyReporter ¶ 

type LatencyReporter func(op Operation, key string, duration time.Duration, err error)

LatencyReporter is a callback invoked after every wire operation with exact execution duration.

type LifecycleState ¶ 

type LifecycleState string

LifecycleState defines the storage lifecycle status of an object. 

const (
	// StatePending indicates that the upload is initiated or pending validation.
	StatePending LifecycleState = "pending"

	// StateCommitted indicates that the upload is finalized and verified in storage.
	StateCommitted LifecycleState = "committed"

	// StateAborted indicates that the upload failed or was aborted before commit.
	StateAborted LifecycleState = "aborted"

	// StateDeleted indicates that the object has been soft-deleted or scheduled for purge.
	StateDeleted LifecycleState = "deleted"
)

type ListOptions ¶ 

type ListOptions struct {
	// Prefix filters objects starting with this prefix string.
	Prefix string

	// Delimiter causes keys containing this delimiter to be grouped in CommonPrefixes.
	Delimiter string

	// Cursor is the continuation token for pagination.
	Cursor string

	// Limit specifies the maximum number of items to return.
	Limit int
}

ListOptions specifies parameters for listing objects in a bucket or prefix.

type ListResult ¶ 

type ListResult struct {
	// Objects is the list of objects returned.
	Objects []Object `json:"objects"`

	// NextCursor is the continuation token to pass to subsequent List calls, or empty if done.
	NextCursor string `json:"next_cursor,omitempty"`

	// CommonPrefixes contains prefixes grouped by the delimiter (e.g. directories).
	CommonPrefixes []string `json:"common_prefixes,omitempty"`

	// IsTruncated indicates whether more items remain in the bucket.
	IsTruncated bool `json:"is_truncated"`
}

ListResult holds the outcome of a List operation.

type MetadataStore ¶ 

type MetadataStore interface {
	// Save creates or updates an object record in the registry.
	Save(ctx context.Context, record *Record) error

	// GetByID retrieves a record by its logical ObjectID.
	GetByID(ctx context.Context, objectID string) (*Record, error)

	// GetByKey retrieves a record by its physical storage key.
	GetByKey(ctx context.Context, key string) (*Record, error)

	// Find queries records matching filter criteria.
	Find(ctx context.Context, filter Filter) ([]Record, error)

	// UpdateStatus transitions an object lifecycle state (e.g. pending -> committed, or soft deleted).
	UpdateStatus(ctx context.Context, objectID string, status LifecycleState) error

	// UpdateMetadata replaces or updates the user-defined metadata map for an object.
	UpdateMetadata(ctx context.Context, objectID string, metadata map[string]string) error

	// UpdateFilename updates the client-facing filename of an object without moving physical bytes.
	UpdateFilename(ctx context.Context, objectID string, newFilename string) error

	// Delete marks an object as soft-deleted or removes it.
	Delete(ctx context.Context, objectID string) error

	// HardDelete permanently purges the record from the database registry.
	HardDelete(ctx context.Context, objectID string) error

	// FindExpired returns committed objects whose ExpiresAt is on or before the given timestamp.
	FindExpired(ctx context.Context, before time.Time, limit int) ([]Record, error)

	// SaveSession persists or updates a resumable multipart upload session.
	SaveSession(ctx context.Context, session *UploadSession) error

	// GetSession retrieves an active upload session by ID.
	GetSession(ctx context.Context, sessionID string) (*UploadSession, error)

	// DeleteSession removes a session upon completion or abort.
	DeleteSession(ctx context.Context, sessionID string) error

	// FindStaleSessions returns active sessions whose ExpiresAt is on or before the given timestamp.
	FindStaleSessions(ctx context.Context, before time.Time, limit int) ([]UploadSession, error)

	// Close gracefully closes any open database pool connections.
	Close() error
}

MetadataStore defines the database metadata repository interface. When enabled, it acts as the source of truth for metadata and indexing.

type NoopObserver ¶ 

type NoopObserver struct{}

NoopObserver is a zero-cost inert observer implementation used when no telemetry is attached.

func (NoopObserver) OnBytesTransferred ¶ 

func (NoopObserver) OnBytesTransferred(op Operation, bytes int64)

func (NoopObserver) OnLimiterWait ¶ 

func (NoopObserver) OnLimiterWait(driver string, waitDuration time.Duration)

func (NoopObserver) OnOperationEnd ¶ 

func (NoopObserver) OnOperationEnd(ctx context.Context, op Operation, key string, duration time.Duration, err error)

func (NoopObserver) OnOperationStart ¶ 

func (NoopObserver) OnOperationStart(ctx context.Context, op Operation, key string) context.Context

type Object ¶ 

type Object struct {
	// ID is the logical identifier (typically UUIDv7) used for database indexing and reference.
	ID string `json:"id"`

	// Namespace partitions objects logically (e.g. "avatars", "bots/autorespon", "media").
	Namespace string `json:"namespace"`

	// OwnerID identifies the tenant, user, or entity owning this object.
	OwnerID string `json:"owner_id,omitempty"`

	// Key is the physical storage path within the provider's bucket.
	Key string `json:"key"`

	// Bucket is the name of the bucket where the object resides.
	Bucket string `json:"bucket"`

	// Size is the object content length in bytes.
	Size int64 `json:"size"`

	// ContentType is the detected or explicitly specified MIME type.
	ContentType string `json:"content_type"`

	// ETag is the entity tag assigned by the storage provider.
	ETag string `json:"etag,omitempty"`

	// ChecksumSHA256 is the hex-encoded SHA-256 hash of the object payload (if computed).
	ChecksumSHA256 string `json:"checksum_sha256,omitempty"`

	// OriginalFilename is the client-facing filename (e.g. "photo.jpg").
	OriginalFilename string `json:"original_filename,omitempty"`

	// Visibility determines whether the object is public or private.
	Visibility Visibility `json:"visibility"`

	// Status indicates the current lifecycle state (pending, committed, deleted).
	Status LifecycleState `json:"status"`

	// Metadata contains user-defined metadata key-value pairs.
	Metadata map[string]string `json:"metadata,omitempty"`

	// RetentionUntil specifies when retention expires. The object cannot be deleted before this date.
	RetentionUntil *time.Time `json:"retention_until,omitempty"`

	// ExpiresAt specifies when the object expires and is eligible for garbage collection.
	ExpiresAt *time.Time `json:"expires_at,omitempty"`

	// LegalHold prevents deletion or modification when true, regardless of retention expiration.
	LegalHold bool `json:"legal_hold,omitempty"`

	// DeletedAt records when the object was soft-deleted.
	DeletedAt *time.Time `json:"deleted_at,omitempty"`

	// ClientChecksum is the optional client-declared payload checksum.
	ClientChecksum string `json:"client_checksum,omitempty"`

	// CreatedAt is the logical creation timestamp.
	CreatedAt time.Time `json:"created_at"`

	// UpdatedAt is the last modification timestamp.
	UpdatedAt time.Time `json:"updated_at"`

	// Provider is the identifier of the driver that stored this object (e.g. "r2-primary", "s3-backup").
	Provider string `json:"provider"`
}

Object represents a stored blob and its canonical system and custom metadata.

type ObjectReader ¶ 

type ObjectReader struct {
	Object
	Body io.ReadCloser
}

ObjectReader pairs object metadata with an active streaming body. Callers must call Close() when finished reading.

func (*ObjectReader) Close ¶ 

func (r *ObjectReader) Close() error

Close closes the underlying reader stream.

func (*ObjectReader) Read ¶ 

func (r *ObjectReader) Read(p []byte) (n int, err error)

Read implements io.Reader by delegating to the underlying body stream.

func (*ObjectReader) WriteTo ¶ 

func (r *ObjectReader) WriteTo(w io.Writer) (int64, error)

WriteTo implements io.WriterTo by forwarding the stream directly to w. If the underlying Body implements io.WriterTo (such as *os.File), it enables zero-copy kernel transfer mechanisms (e.g. Linux sendfile(2) / splice(2)) in standard HTTP handlers.

type Observer ¶ 

type Observer interface {
	// OnOperationStart is invoked at the beginning of an object storage operation.
	// It may enrich and return a child context (e.g. for distributed tracing spans).
	OnOperationStart(ctx context.Context, op Operation, key string) context.Context

	// OnOperationEnd is invoked when an operation finishes, reporting duration and error status.
	OnOperationEnd(ctx context.Context, op Operation, key string, duration time.Duration, err error)

	// OnBytesTransferred records data transfer volume for streaming uploads and downloads.
	OnBytesTransferred(op Operation, bytes int64)

	// OnLimiterWait records backpressure delay caused by concurrency rate limiters.
	OnLimiterWait(driver string, waitDuration time.Duration)
}

Observer provides vendor-neutral instrumentation hooks for telemetry, metrics collection, distributed tracing, and audit logging.

type Operation ¶ 

type Operation string

Operation represents the high-level storage action being performed. 

const (
	OpPut             Operation = "put"
	OpGet             Operation = "get"
	OpHead            Operation = "head"
	OpDelete          Operation = "delete"
	OpList            Operation = "list"
	OpPresignGet      Operation = "presign_get"
	OpPresignPut      Operation = "presign_put"
	OpCopy            Operation = "copy"
	OpMove            Operation = "move"
	OpUploadPart      Operation = "upload_part"
	OpMultipart       Operation = "multipart"
	OpSoftDelete      Operation = "soft_delete"
	OpRestore         Operation = "restore"
	OpPermanentDelete Operation = "permanent_delete"
)

type Option ¶ 

type Option func(*Client)

Option represents a functional configuration option for the BlobKit Client.

func WithCache ¶ 

func WithCache(cache Cache) Option

WithCache sets an optional caching layer for object metadata and negative lookups.

func WithDefaultVisibility ¶ 

func WithDefaultVisibility(vis Visibility) Option

WithDefaultVisibility sets the default visibility for uploaded objects.

func WithDriver ¶ 

func WithDriver(driver Driver) Option

WithDriver sets a single fixed storage driver for the client.

func WithKeyGenerator ¶ 

func WithKeyGenerator(gen key.Generator) Option

WithKeyGenerator configures the physical key naming strategy.

func WithObserver ¶ 

func WithObserver(obs Observer) Option

WithObserver sets a vendor-neutral observability instrumentation hook.

func WithPolicy ¶ 

func WithPolicy(p Policy) Option

WithPolicy sets the default upload security and validation policy for the client.

func WithRegistry ¶ 

func WithRegistry(reg MetadataStore) Option

WithRegistry attaches an optional database metadata registry.

func WithRouter ¶ 

func WithRouter(r Router) Option

WithRouter sets a custom routing policy (failover, namespace, etc.).

func WithURLResolver ¶ 

func WithURLResolver(resolver URLResolver) Option

WithURLResolver configures the public delivery URL resolver.

type Policy ¶ 

type Policy struct {
	// AllowedMIMEs is a whitelist of permitted MIME types (e.g. ["image/jpeg", "image/*", "audio/*"]).
	AllowedMIMEs []string

	// DeniedMIMEs is a blacklist of forbidden MIME types (e.g. ["application/x-dosexec", "application/x-sh"]).
	DeniedMIMEs []string

	// AllowedExtensions is a whitelist of permitted file extensions (e.g. [".jpg", ".png", ".pdf"]).
	AllowedExtensions []string

	// DeniedExtensions is a blacklist of forbidden extensions (e.g. [".exe", ".sh", ".bat", ".cmd", ".php"]).
	DeniedExtensions []string

	// MaxObjectSize is the maximum allowed payload size in bytes (0 means unlimited).
	MaxObjectSize int64

	// MinObjectSize is the minimum required payload size in bytes (0 means no minimum).
	MinObjectSize int64

	// MaxFilenameLength is the maximum allowed filename length (defaults to 255).
	MaxFilenameLength int

	// DisallowMIMEMismatch flags whether an obvious mismatch between file extension and detected MIME is forbidden.
	DisallowMIMEMismatch bool

	// ContentScanner is an optional custom scanner hook (e.g. antivirus or deep payload inspection).
	// It MUST return an io.Reader that allows subsequent operations to read the full payload,
	// either by returning the original reader unconsumed, or by returning a new reader that buffers/replays the consumed bytes.
	ContentScanner func(ctx context.Context, r io.Reader, filename, mime string) (io.Reader, error)
}

Policy enforces upload security constraints, format whitelists, and size boundaries.

func (*Policy) Validate ¶ 

func (p *Policy) Validate(ctx context.Context, in ValidationInput) (io.Reader, error)

Validate executes policy rules against the given input. It returns a potentially replacement io.Reader if the ContentScanner consumes bytes.

type PresignOptions ¶ 

type PresignOptions struct {
	// Expiry is the validity duration of the presigned URL. Default is 15 minutes.
	Expiry time.Duration

	// ContentType specifies the expected Content-Type header to constrain PUT presigned URLs.
	ContentType string

	// ContentDisposition specifies the expected Content-Disposition header.
	ContentDisposition string

	// QueryParams allows passing additional query parameters if supported.
	QueryParams url.Values
}

PresignOptions configures parameters for URL presigning.

type PresignedURL ¶ 

type PresignedURL struct {
	// URL is the signed HTTP address.
	URL string `json:"url"`

	// Method is the expected HTTP verb (e.g. "GET" or "PUT").
	Method string `json:"method"`

	// ExpiresAt is the timestamp when the signature becomes invalid.
	ExpiresAt time.Time `json:"expires_at"`

	// SignedHeaders contains any required HTTP headers that were included in the signature calculation.
	SignedHeaders map[string]string `json:"signed_headers,omitempty"`
}

PresignedURL contains a time-limited signed URL for direct client access or uploads.

type PutOptions ¶ 

type PutOptions struct {
	// ID is an optional predetermined logical ID. If empty, a UUIDv7 is generated automatically.
	ID string

	// Key is an optional explicit physical storage key. If empty, KeyGenerator determines it.
	Key string

	// Namespace partitions objects logically (e.g. "avatars", "bots/autorespon", "media").
	Namespace string

	// OwnerID identifies the owner/tenant of the object.
	OwnerID string

	// Filename is the original client filename (e.g. "report.pdf"), used for Content-Disposition & MIME hints.
	Filename string

	// ContentType is the MIME type. If empty, BlobKit sniffs up to the first 512 bytes.
	ContentType string

	// Size is the known payload size in bytes. Set to SizeUnknown (-1) if unknown or streaming.
	Size int64

	// ExplicitSize indicates that Size was intentionally specified (even if 0),
	// allowing distinction between an unassigned size (streaming) and an explicit 0-byte upload.
	ExplicitSize bool

	// Visibility controls whether the object is public or private. Defaults to VisibilityPrivate.
	Visibility Visibility

	// Metadata contains custom user key-value metadata to store alongside the object.
	Metadata map[string]string

	// ContentDisposition customizes the Content-Disposition HTTP header.
	ContentDisposition string

	// CacheControl customizes the Cache-Control HTTP header (e.g. "public, max-age=31536000, immutable").
	CacheControl string

	// RetentionUntil locks the object against deletion or modification until this timestamp.
	RetentionUntil *time.Time

	// ExpiresAt marks the object to expire automatically after this timestamp.
	ExpiresAt *time.Time

	// LegalHold marks the object with an active legal hold.
	LegalHold bool

	// ClientChecksum specifies an expected payload SHA-256 hash to enforce on upload.
	ClientChecksum string

	// VerifyIntegrity forces BlobKit to stream-hash and verify the payload SHA-256 during upload.
	VerifyIntegrity bool

	// Policy overrides client-wide upload validation rules for this specific upload.
	Policy *Policy

	// Provider forces the upload to use a specific registered driver instead of the router policy.
	Provider string
}

PutOptions configures an upload operation, conveying semantic intent.

type ReconciliationReport ¶ 

type ReconciliationReport struct {
	TotalChecked   int64    `json:"total_checked"`
	OrphanObjects  []string `json:"orphan_objects"`  // present in storage but missing in registry
	GhostRecords   []string `json:"ghost_records"`   // present in registry (StateCommitted) but missing in storage
	MismatchedSize []string `json:"mismatched_size"` // length mismatch between registry and storage
	Repaired       int64    `json:"repaired"`        // records or objects repaired (when dryRun is false)
}

ReconciliationReport records orphan objects, ghost records, and repaired inconsistencies.

type Record ¶ 

type Record struct {
	// ObjectID is the immutable logical identifier (UUIDv7) and primary database key.
	ObjectID string `json:"object_id"`

	// Namespace partitions objects logically (e.g. "avatars", "bots/autorespon", "media").
	Namespace string `json:"namespace"`

	// OwnerID identifies the tenant, user, or bot owning this object.
	OwnerID string `json:"owner_id,omitempty"`

	// Key is the physical storage key inside the provider's bucket.
	Key string `json:"key"`

	// Bucket is the storage bucket name.
	Bucket string `json:"bucket"`

	// Provider is the identifier of the driver holding the bytes (e.g. "r2-primary").
	Provider string `json:"provider"`

	// MIMEType is the verified Content-Type of the blob.
	MIMEType string `json:"mime_type"`

	// Size is the content length in bytes.
	Size int64 `json:"size"`

	// ChecksumSHA256 is the hex-encoded SHA-256 payload digest.
	ChecksumSHA256 string `json:"checksum_sha256,omitempty"`

	// ETag is the HTTP entity tag from the storage provider.
	ETag string `json:"etag,omitempty"`

	// OriginalFilename is the client-facing filename (e.g. "photo.png").
	OriginalFilename string `json:"original_filename,omitempty"`

	// Visibility specifies access level (public or private).
	Visibility Visibility `json:"visibility"`

	// Status indicates lifecycle state (pending, committed, deleted).
	Status LifecycleState `json:"status"`

	// Metadata holds arbitrary user-defined key-value attributes.
	Metadata map[string]string `json:"metadata,omitempty"`

	// RetentionUntil specifies when retention lock expires.
	RetentionUntil *time.Time `json:"retention_until,omitempty"`

	// ExpiresAt specifies when the object automatically expires.
	ExpiresAt *time.Time `json:"expires_at,omitempty"`

	// LegalHold prevents deletion or modification when true.
	LegalHold bool `json:"legal_hold,omitempty"`

	// ClientChecksum is the client-provided checksum (if any).
	ClientChecksum string `json:"client_checksum,omitempty"`

	// CreatedAt is the logical creation timestamp.
	CreatedAt time.Time `json:"created_at"`

	// UpdatedAt is the last modification timestamp.
	UpdatedAt time.Time `json:"updated_at"`

	// DeletedAt records when the object was marked deleted (if soft-deleted).
	DeletedAt *time.Time `json:"deleted_at,omitempty"`
}

Record represents the canonical metadata and indexing record stored in the database registry. NOTE: Full URL is explicitly omitted from canonical record storage, as URLs are delivery/access details dynamically constructed by BlobKit when requested (CDN or presigned).

func (*Record) ToObject ¶ 

func (r *Record) ToObject() Object

ToObject converts a registry Record into a BlobKit Object reference.

type RouteContext ¶ 

type RouteContext struct {
	// Op is the current operation.
	Op Operation

	// Key is the target physical storage key (if known).
	Key string

	// Namespace is the semantic category/partition (e.g. "avatars", "backups").
	Namespace string

	// ForcedProvider is an optional explicit provider name requested by the caller.
	ForcedProvider string
}

RouteContext carries routing intent from the application or BlobKit core to the router.

type Router ¶ 

type Router interface {
	// Select resolves the appropriate driver for the given operation context.
	Select(ctx context.Context, rc RouteContext) (Driver, error)

	// AllDrivers returns all registered drivers under this router.
	AllDrivers() []Driver

	// ReportFailure notifies the router of an execution failure for health tracking.
	ReportFailure(driverName string, err error)

	// ReportSuccess notifies the router of a successful execution.
	ReportSuccess(driverName string)
}

Router determines which storage provider driver should execute an operation.

type S3PathResolver ¶ 

type S3PathResolver struct {
	// contains filtered or unexported fields
}

S3PathResolver formats standard path-style or virtual-hosted URLs for S3 endpoints.

func NewS3PathResolver ¶ 

func NewS3PathResolver(endpoint, bucket string) *S3PathResolver

NewS3PathResolver returns an S3PathResolver.

func (*S3PathResolver) ResolveURL ¶ 

func (s *S3PathResolver) ResolveURL(key string) string

ResolveURL builds a path-style URL: endpoint/bucket/key.

type SeekableReader ¶ 

type SeekableReader struct {
	// contains filtered or unexported fields
}

SeekableReader implements io.ReadSeekCloser and io.ReaderAt over an object storage key using HTTP Byte-Range requests ("Range: bytes=start-end").

It enables fast, on-demand random-access seeking (e.g. for ZIP parsing, Parquet query engines, audio/video streaming, or PDF reading) without downloading the entire object into memory or disk.

func (*SeekableReader) Close ¶ 

func (s *SeekableReader) Close() error

Close closes any open active stream connections.

func (*SeekableReader) Read ¶ 

func (s *SeekableReader) Read(p []byte) (int, error)

Read reads up to len(p) bytes from the remote object starting at the current cursor.

func (*SeekableReader) ReadAt ¶ 

func (s *SeekableReader) ReadAt(p []byte, off int64) (int, error)

ReadAt reads len(p) bytes into p at offset off without modifying the seeker's cursor. It implements io.ReaderAt and is safe for concurrent execution across goroutines.

func (*SeekableReader) Seek ¶ 

func (s *SeekableReader) Seek(offset int64, whence int) (int64, error)

Seek sets the offset for the next Read, adhering to the standard io.Seeker contract.

func (*SeekableReader) Size ¶ 

func (s *SeekableReader) Size() int64

Size returns the total size of the remote object in bytes.

type SessionState ¶ 

type SessionState string

SessionState represents the lifecycle status of a resumable multipart upload session. 

const (
	// SessionActive indicates the upload session is open and receiving parts.
	SessionActive SessionState = "active"

	// SessionCommitted indicates the upload session has been finalized into an object.
	SessionCommitted SessionState = "committed"

	// SessionAborted indicates the upload session was cancelled or cleaned up.
	SessionAborted SessionState = "aborted"
)

type SizeReader ¶ 

type SizeReader struct {
	// contains filtered or unexported fields
}

SizeReader wraps an io.Reader to enforce expected payload size constraints. If exact is true (or expectedSize >= 0), it returns ErrSizeMismatch if the stream yields fewer bytes than expected or has surplus bytes beyond expectedSize.

func NewSizeReader ¶ 

func NewSizeReader(r io.Reader, expectedSize int64, exact bool) *SizeReader

NewSizeReader constructs a SizeReader. If exact is true and expectedSize >= 0, the reader strictly verifies byte parity. If expectedSize < 0, size verification is disabled (pass-through).

func (*SizeReader) Read ¶ 

func (sr *SizeReader) Read(p []byte) (n int, err error)

Read implements io.Reader with strict boundary checking.

func (*SizeReader) TotalRead ¶ 

func (sr *SizeReader) TotalRead() int64

TotalRead returns the total number of bytes read so far.

func (*SizeReader) Verify ¶ 

func (sr *SizeReader) Verify() error

Verify checks whether the reader was read to completion and matched expectedSize.

type StorageError ¶ 

type StorageError struct {
	Op       string
	Key      string
	Provider string
	Err      error
}

StorageError represents a standardized, sanitized error returned by BlobKit operations. It wraps the underlying driver error without leaking sensitive endpoint credentials or internal network topologies.

func (*StorageError) Error ¶ 

func (e *StorageError) Error() string

Error formats the sanitized error message.

func (*StorageError) Is ¶ 

func (e *StorageError) Is(target error) bool

Is reports whether the target matches this error or its wrapped error.

func (*StorageError) Unwrap ¶ 

func (e *StorageError) Unwrap() error

Unwrap returns the underlying error.

type URLResolver ¶ 

type URLResolver interface {
	ResolveURL(key string) string
}

URLResolver maps a storage object key to an accessible public or CDN URL.

type UploadSession ¶ 

type UploadSession struct {
	// ID is the unique identifier of the session (UUIDv7).
	ID string `json:"id"`

	// ObjectID is the logical object ID being constructed.
	ObjectID string `json:"object_id"`

	// Key is the physical storage key in the bucket.
	Key string `json:"key"`

	// Bucket is the target bucket name.
	Bucket string `json:"bucket"`

	// Provider is the storage driver managing the multipart upload.
	Provider string `json:"provider"`

	// UploadID is the underlying driver's multipart upload identifier.
	UploadID string `json:"upload_id"`

	// PartSize is the planned byte size per chunk (typically >= 5MB).
	PartSize int64 `json:"part_size"`

	// TotalSize is the optional declared total object size, or 0 if unknown.
	TotalSize int64 `json:"total_size"`

	// Parts tracks all successfully uploaded parts for this session.
	Parts []CompletedPart `json:"parts"`

	// ExpiresAt specifies when this session becomes stale and eligible for auto-cleanup.
	ExpiresAt time.Time `json:"expires_at"`

	// CreatedAt is the timestamp when the session was initiated.
	CreatedAt time.Time `json:"created_at"`

	// Status indicates whether the session is active, committed, or aborted.
	Status SessionState `json:"status"`
}

UploadSession encapsulates the state required to track, resume, or abort a multipart upload.

type ValidationInput ¶ 

type ValidationInput struct {
	Namespace    string
	Filename     string
	ContentType  string
	Size         int64
	ExplicitSize bool // ExplicitSize indicates that Size was explicitly specified (even if 0)
	SizeKnown    bool // SizeKnown indicates that the payload size is deterministically known
	StreamReader io.Reader
}

ValidationInput supplies parameters for policy enforcement.

type Visibility ¶ 

type Visibility string

Visibility defines the access control scope of an object. 

const (
	// VisibilityPublic marks the object as publicly deliverable via CDN or bucket public endpoint.
	VisibilityPublic Visibility = "public"

	// VisibilityPrivate marks the object as private, requiring presigned URLs or authenticated streams.
	VisibilityPrivate Visibility = "private"
)

Source Files ¶

View all Source files

Directories ¶

Path Synopsis
provider
azure
fs
gcs
gdrive
memory
s3
sftp
webdav
postgres
sql
sqlite
fleet

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.