store

package
v2.2.0 Latest Latest
Warning

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

 Go to latest Published: Aug 12, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Overview

Package store provides a generic, typed key-value storage layer used by workflow tasks to persist and retrieve intermediate data.

The two main interfaces are:

  • RawStore — byte-level storage with Upload, Download, Delete, Exists, and List operations. Concrete implementations include LocalStore (filesystem) and S3Store (S3-compatible object storage).

  • Store — a typed wrapper around RawStore that uses a Codec to serialize and deserialize Go values of any type T. JSONCodec is the default codec.

Keys are built with the KeyBuilder helper to ensure consistent, hierarchical naming across stores. The InstrumentedStore decorator adds OpenTelemetry tracing and metrics to any RawStore implementation.

Index

Constants

View Source 
const (
	// FileTypeDirectory represents a directory artifact type.
	FileTypeDirectory = "directory"
	// FileTypeFile represents a single-file artifact type.
	FileTypeFile = "file"
	// FileTypeArchive represents a tar.gz archive artifact type.
	FileTypeArchive = "archive"
)
View Source 
const (
	// MaxUploadSize is the maximum size for uploads (1GB).
	MaxUploadSize = 1 << 30
)

Variables

This section is empty.

Functions

func ArchiveDirectory 

func ArchiveDirectory(sourceDir string, writer io.Writer) (err error)

ArchiveDirectory creates a tar.gz archive of a directory.

func DeletePrefix 

func DeletePrefix(ctx context.Context, raw RawStore, prefix string) error

DeletePrefix removes every key stored under the given prefix. It lists the prefix and deletes each key, continuing past individual failures and reporting the first error encountered.

func DownloadFile 

func DownloadFile(ctx context.Context, raw RawStore, key, destPath, typ string) (err error)

DownloadFile downloads the data stored under key to a local path. Supported types: "file" writes the file (creating parent directories); "directory"/"archive" extracts a tar.gz stream into destPath.

func ExtractArchive 

func ExtractArchive(reader io.Reader, destDir string) error

ExtractArchive extracts a tar.gz archive to a directory.

func UploadFile 

func UploadFile(ctx context.Context, raw RawStore, key, sourcePath, typ string) error

UploadFile uploads a file or directory from the local filesystem to the store under the given key. Supported types: "file" uploads the file as-is; "directory"/"archive" stream a tar.gz of the directory via io.Pipe. An empty typ auto-detects from the source path. Uploads are capped at MaxUploadSize (enforced by RawStore implementations).

Types

type BytesCodec 

type BytesCodec struct{}

BytesCodec is a pass-through codec for []byte values.

func (*BytesCodec) Decode 

func (c *BytesCodec) Decode(reader io.Reader) ([]byte, error)

Decode reads all bytes from the reader.

func (*BytesCodec) Encode 

func (c *BytesCodec) Encode(value []byte) (io.Reader, error)

Encode wraps the byte slice in a reader.

type Codec 

type Codec[T any] interface {
	// Encode serializes a value into a reader.
	Encode(value T) (io.Reader, error)

	// Decode deserializes a value from a reader.
	Decode(reader io.Reader) (T, error)
}

Codec defines a serialization strategy for type T.

type InstrumentedStore 

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

InstrumentedStore wraps any RawStore with OpenTelemetry spans and metrics. When OTel config is not in context, all calls delegate directly to the inner store with zero overhead.

func (*InstrumentedStore) Close 

func (s *InstrumentedStore) Close() error

func (*InstrumentedStore) Delete 

func (s *InstrumentedStore) Delete(ctx context.Context, key string) error

func (*InstrumentedStore) Download 

func (s *InstrumentedStore) Download(ctx context.Context, key string) (io.ReadCloser, error)

func (*InstrumentedStore) Exists 

func (s *InstrumentedStore) Exists(ctx context.Context, key string) (bool, error)

func (*InstrumentedStore) List 

func (s *InstrumentedStore) List(ctx context.Context, prefix string) ([]string, error)

func (*InstrumentedStore) Upload 

func (s *InstrumentedStore) Upload(ctx context.Context, key string, data io.Reader) error

type JSONCodec 

type JSONCodec[T any] struct{}

JSONCodec serializes and deserializes values as JSON.

func (*JSONCodec[T]) Decode 

func (c *JSONCodec[T]) Decode(reader io.Reader) (T, error)

Decode deserializes a value from JSON.

func (*JSONCodec[T]) Encode 

func (c *JSONCodec[T]) Encode(value T) (io.Reader, error)

Encode serializes a value to JSON.

type KeyBuilder 

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

KeyBuilder provides composable key generation for storage keys. Keys are built by appending path segments and joining them with slashes. KeyBuilder is immutable — each With* method returns a new instance, making it safe to branch from a shared base.

func NewKeyBuilder 

func NewKeyBuilder() *KeyBuilder

NewKeyBuilder creates a new empty KeyBuilder.

func (*KeyBuilder) Build 

func (kb *KeyBuilder) Build() string

Build joins all segments with slashes and returns the key.

func (*KeyBuilder) WithName 

func (kb *KeyBuilder) WithName(name string) *KeyBuilder

WithName appends a name segment.

func (*KeyBuilder) WithRun 

func (kb *KeyBuilder) WithRun(id string) *KeyBuilder

WithRun appends a run ID segment.

func (*KeyBuilder) WithStep 

func (kb *KeyBuilder) WithStep(name string) *KeyBuilder

WithStep appends a step name segment.

func (*KeyBuilder) WithWorkflow 

func (kb *KeyBuilder) WithWorkflow(id string) *KeyBuilder

WithWorkflow appends a workflow ID segment.

type LocalStore 

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

LocalStore implements RawStore using the local filesystem.

func NewLocalStore 

func NewLocalStore(basePath string) (*LocalStore, error)

NewLocalStore creates a new LocalStore rooted at basePath. The base directory is created if it does not exist.

func (*LocalStore) Close 

func (s *LocalStore) Close() error

Close is a no-op for LocalStore.

func (*LocalStore) Delete 

func (s *LocalStore) Delete(_ context.Context, key string) error

Delete removes the data stored under the given key.

func (*LocalStore) Download 

func (s *LocalStore) Download(_ context.Context, key string) (io.ReadCloser, error)

Download retrieves data for the given key from the local filesystem.

func (*LocalStore) Exists 

func (s *LocalStore) Exists(_ context.Context, key string) (bool, error)

Exists checks whether data exists under the given key.

func (*LocalStore) List 

func (s *LocalStore) List(_ context.Context, prefix string) ([]string, error)

List returns all keys matching the given prefix.

func (*LocalStore) Upload 

func (s *LocalStore) Upload(_ context.Context, key string, data io.Reader) error

Upload stores data under the given key on the local filesystem.

type RawStore 

type RawStore interface {
	// Upload stores data under the given key.
	Upload(ctx context.Context, key string, data io.Reader) error

	// Download retrieves data for the given key.
	// The caller must close the returned ReadCloser.
	Download(ctx context.Context, key string) (io.ReadCloser, error)

	// Delete removes the data stored under the given key.
	Delete(ctx context.Context, key string) error

	// Exists checks whether data exists under the given key.
	Exists(ctx context.Context, key string) (bool, error)

	// List returns all keys matching the given prefix.
	List(ctx context.Context, prefix string) ([]string, error)

	// Close releases any resources held by the store.
	Close() error
}

RawStore is a byte-level storage interface. Implementations handle raw byte persistence with string keys.

func NewInstrumentedStore 

func NewInstrumentedStore(inner RawStore) RawStore

NewInstrumentedStore creates a new InstrumentedStore wrapping the given store.

func NewS3Store 

func NewS3Store(ctx context.Context, cfg S3Config) (RawStore, error)

NewS3Store creates a new S3 raw store. It auto-creates the bucket if it does not exist.

type S3Config 

type S3Config struct {
	// Endpoint is the S3-compatible server endpoint (e.g., "localhost:9000")
	Endpoint string

	// AccessKey is the access key ID
	AccessKey string

	// SecretKey is the secret access key
	SecretKey string

	// Bucket is the bucket name for storing data
	Bucket string

	// Prefix is an optional prefix for all object keys
	Prefix string

	// UseSSL determines whether to use HTTPS
	UseSSL bool

	// Region is the bucket region (defaults to "us-east-1" if empty)
	Region string
}

S3Config contains configuration for connecting to an S3-compatible storage service.

type S3Store 

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

S3Store implements RawStore using S3-compatible storage via AWS SDK v2.

func (*S3Store) Close 

func (s *S3Store) Close() error

Close releases any resources held by the store (no-op for S3).

func (*S3Store) Delete 

func (s *S3Store) Delete(ctx context.Context, key string) error

Delete removes the data stored under the given key.

func (*S3Store) Download 

func (s *S3Store) Download(ctx context.Context, key string) (io.ReadCloser, error)

Download retrieves data for the given key. The caller must close the returned ReadCloser.

func (*S3Store) Exists 

func (s *S3Store) Exists(ctx context.Context, key string) (bool, error)

Exists checks whether data exists under the given key.

func (*S3Store) List 

func (s *S3Store) List(ctx context.Context, prefix string) ([]string, error)

List returns all keys matching the given prefix. The returned keys have the store's prefix stripped.

func (*S3Store) Upload 

func (s *S3Store) Upload(ctx context.Context, key string, data io.Reader) error

Upload stores data under the given key.

type Store 

type Store[T any] interface {
	// Save serializes and stores the value under the given key.
	Save(ctx context.Context, key string, value T) error

	// Load retrieves and deserializes the value stored under the given key.
	Load(ctx context.Context, key string) (T, error)

	// Delete removes the data stored under the given key.
	Delete(ctx context.Context, key string) error

	// Exists checks whether data exists under the given key.
	Exists(ctx context.Context, key string) (bool, error)

	// List returns all keys matching the given prefix.
	List(ctx context.Context, prefix string) ([]string, error)

	// Close releases any resources held by the store.
	Close() error
}

Store is a typed storage interface with automatic serialization.

func NewBytesStore 

func NewBytesStore(raw RawStore) Store[[]byte]

NewBytesStore creates a Store[[]byte] that passes through raw bytes.

func NewJSONStore 

func NewJSONStore[T any](raw RawStore) Store[T]

NewJSONStore creates a Store[T] that serializes values as JSON.

func NewTypedStore 

func NewTypedStore[T any](raw RawStore, codec Codec[T]) Store[T]

NewTypedStore creates a new TypedStore from a RawStore and Codec.

type TypedStore 

type TypedStore[T any] struct {
	// contains filtered or unexported fields
}

TypedStore adapts a RawStore with a Codec to implement Store[T].

func (*TypedStore[T]) Close 

func (s *TypedStore[T]) Close() error

Close releases any resources held by the underlying RawStore.

func (*TypedStore[T]) Delete 

func (s *TypedStore[T]) Delete(ctx context.Context, key string) error

Delete removes the data stored under the given key.

func (*TypedStore[T]) Exists 

func (s *TypedStore[T]) Exists(ctx context.Context, key string) (bool, error)

Exists checks whether data exists under the given key.

func (*TypedStore[T]) List 

func (s *TypedStore[T]) List(ctx context.Context, prefix string) ([]string, error)

List returns all keys matching the given prefix.

func (*TypedStore[T]) Load 

func (s *TypedStore[T]) Load(ctx context.Context, key string) (T, error)

Load downloads data from the underlying RawStore and deserializes it.

func (*TypedStore[T]) Save 

func (s *TypedStore[T]) Save(ctx context.Context, key string, value T) error

Save serializes the value and uploads it to the underlying RawStore.

Source Files

View all Source files

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.