storage

package
v1.5.4 Latest Latest
Warning

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

Go to latest
Published: Jul 2, 2026 License: Apache-2.0 Imports: 9 Imported by: 2

Documentation

Overview

Package storage exposes the MediaStorageService interface for media persistence, plus instrumentation that wires storage operations into the runtime's event bus and OpenTelemetry spans.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type InstrumentedStorage added in v1.4.7

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

InstrumentedStorage wraps any MediaStorageService with telemetry — OTel spans on every call, runtime events on success/failure, and (when wired in by the consumer) bytes/latency/error metrics derived from the emitted events.

The wrapper is nil-safe at every level: a nil bus or nil inner is a hard error, but missing fields on MediaMetadata never panic.

Backend identification (`local`, `s3`, ...) is supplied by the caller at construction time so the same wrapper works regardless of the concrete inner implementation.

func NewInstrumentedStorage added in v1.4.7

func NewInstrumentedStorage(inner MediaStorageService, bus events.Bus, backend string) *InstrumentedStorage

NewInstrumentedStorage wraps inner so every call publishes a media lifecycle event to bus and records an OTel span. backend names the concrete implementation (e.g. "local", "s3") and ends up on every emitted event and span.

Passing a nil inner returns nil — the wrapper has nothing to wrap. A nil bus is allowed (events become no-ops); spans still emit via the global tracer provider.

func (*InstrumentedStorage) DeleteMedia added in v1.4.7

func (s *InstrumentedStorage) DeleteMedia(ctx context.Context, reference Reference) error

DeleteMedia traces, times, and announces a delete operation.

func (*InstrumentedStorage) GetURL added in v1.4.7

func (s *InstrumentedStorage) GetURL(
	ctx context.Context, reference Reference, expiry time.Duration,
) (string, error)

GetURL traces and times URL generation. No success event is emitted — URL generation is a metadata operation that doesn't move bytes — but failures still publish an error event so opaque 404s are visible.

func (*InstrumentedStorage) Inner added in v1.4.7

Inner returns the wrapped MediaStorageService — useful for callers that need to type-assert to a backend-specific interface.

func (*InstrumentedStorage) RetrieveMedia added in v1.4.7

func (s *InstrumentedStorage) RetrieveMedia(
	ctx context.Context, reference Reference,
) (*types.MediaContent, error)

RetrieveMedia traces, times, and announces a retrieve operation.

func (*InstrumentedStorage) SetBus added in v1.4.7

func (s *InstrumentedStorage) SetBus(bus events.Bus)

SetBus swaps the event bus the wrapper publishes to. Arena constructs the storage before the runtime event bus exists, then late-binds it via this method once the bus is wired up. Safe to call once before the wrapper sees any traffic; safe to call concurrently with operations because the bus pointer is guarded by busMu.

func (*InstrumentedStorage) StoreMedia added in v1.4.7

func (s *InstrumentedStorage) StoreMedia(
	ctx context.Context, content *types.MediaContent, metadata *MediaMetadata,
) (Reference, error)

StoreMedia traces, times, and announces a store operation.

type MediaMetadata

type MediaMetadata struct {
	// RunID identifies the test run that generated this media
	RunID string `json:"run_id"`

	// ConversationID identifies the conversation containing this media
	ConversationID string `json:"conversation_id,omitempty"`

	// SessionID identifies the session (for streaming sessions)
	SessionID string `json:"session_id,omitempty"`

	// MessageIdx is the index of the message containing this media (0-based)
	MessageIdx int `json:"message_idx"`

	// PartIdx is the index of the content part containing this media (0-based)
	PartIdx int `json:"part_idx"`

	// MIMEType is the media MIME type (e.g., "image/jpeg", "audio/mp3")
	MIMEType string `json:"mime_type"`

	// SizeBytes is the size of the media content in bytes
	SizeBytes int64 `json:"size_bytes"`

	// ProviderID identifies the provider that generated this media
	ProviderID string `json:"provider_id,omitempty"`

	// Timestamp is when the media was stored
	Timestamp time.Time `json:"timestamp"`

	// PolicyName is the retention policy to apply to this media
	PolicyName string `json:"policy_name,omitempty"`
}

MediaMetadata contains metadata about stored media for organization and policy enforcement. This metadata is used to organize media files in storage and apply retention policies.

type MediaStorageService

type MediaStorageService interface {
	// StoreMedia stores media content and returns a storage reference.
	// The reference can be used to retrieve the media later.
	//
	// Parameters:
	//   - ctx: Context for cancellation and timeouts
	//   - content: The media content to store (must have Data, FilePath, or URL set)
	//   - metadata: Metadata about the media for organization and policies
	//
	// Returns:
	//   - Reference that can be used to retrieve the media
	//   - Error if storage fails
	//
	// The implementation should:
	//   - Validate the content and metadata
	//   - Store the media content durably
	//   - Apply any configured policies (e.g., retention)
	//   - Return a reference that uniquely identifies the stored media
	StoreMedia(ctx context.Context, content *types.MediaContent, metadata *MediaMetadata) (Reference, error)

	// RetrieveMedia retrieves media content by its storage reference.
	//
	// Parameters:
	//   - ctx: Context for cancellation and timeouts
	//   - reference: The storage reference returned by StoreMedia
	//
	// Returns:
	//   - MediaContent with FilePath set (Data should NOT be loaded into memory)
	//   - Error if retrieval fails or reference is invalid
	//
	// The implementation should:
	//   - Validate the reference
	//   - Return MediaContent with FilePath pointing to the stored media
	//   - NOT load the full media data into memory (caller can use GetBase64Data if needed)
	RetrieveMedia(ctx context.Context, reference Reference) (*types.MediaContent, error)

	// DeleteMedia deletes media content by its storage reference.
	//
	// Parameters:
	//   - ctx: Context for cancellation and timeouts
	//   - reference: The storage reference to delete
	//
	// Returns:
	//   - Error if deletion fails or reference is invalid
	//
	// The implementation should:
	//   - Validate the reference
	//   - Delete the media content if not referenced elsewhere (for dedup)
	//   - Clean up any associated metadata
	//   - Handle concurrent deletions safely
	DeleteMedia(ctx context.Context, reference Reference) error

	// GetURL returns a URL that can be used to access the media.
	// For local storage, this returns a file:// URL.
	// For cloud storage, this may return a signed URL with expiration.
	//
	// Parameters:
	//   - ctx: Context for cancellation and timeouts
	//   - reference: The storage reference
	//   - expiry: How long the URL should be valid (ignored for local storage)
	//
	// Returns:
	//   - URL string that can be used to access the media
	//   - Error if URL generation fails or reference is invalid
	GetURL(ctx context.Context, reference Reference, expiry time.Duration) (string, error)
}

MediaStorageService defines the interface for storing and retrieving media content. Implementations may store media in local filesystem, cloud storage, or other backends.

Example usage:

storage := local.NewFileStore("/var/promptkit/media")
ref, err := storage.StoreMedia(ctx, mediaContent, metadata)
if err != nil {
    return err
}
// Later...
content, err := storage.RetrieveMedia(ctx, ref)

Implementations should be safe for concurrent use by multiple goroutines.

type OrganizationMode

type OrganizationMode string

OrganizationMode defines how media files are organized in storage.

const (
	// OrganizationBySession organizes media by session ID
	OrganizationBySession OrganizationMode = "by-session"

	// OrganizationByConversation organizes media by conversation ID
	OrganizationByConversation OrganizationMode = "by-conversation"

	// OrganizationByRun organizes media by run ID
	OrganizationByRun OrganizationMode = "by-run"
)

type PolicyHandler

type PolicyHandler interface {
	// ApplyPolicy applies a named policy to a media file.
	// This typically stores policy metadata alongside the media.
	//
	// Parameters:
	//   - ctx: Context for cancellation and timeouts
	//   - filePath: Path to the media file
	//   - policyName: Name of the policy to apply (e.g., "delete-after-10min", "retain-30days")
	//
	// Returns:
	//   - Error if policy application fails or policy is unknown
	ApplyPolicy(ctx context.Context, filePath string, policyName string) error

	// EnforcePolicy scans stored media and enforces policies.
	// This is typically called periodically in the background.
	//
	// Parameters:
	//   - ctx: Context for cancellation and timeouts
	//
	// Returns:
	//   - Error if enforcement fails (should log but not crash on individual file errors)
	//
	// The implementation should:
	//   - Scan media directories for policy metadata
	//   - Apply policies (e.g., delete expired files)
	//   - Log enforcement actions
	//   - Handle errors gracefully (don't stop on permission denied, etc.)
	EnforcePolicy(ctx context.Context) error
}

PolicyHandler defines the interface for applying and enforcing storage policies. Policies control media retention, cleanup, and other lifecycle management.

Example usage:

policy := policy.NewTimeBasedPolicy()
err := policy.ApplyPolicy(ctx, "/path/to/media.jpg", "delete-after-10min")
if err != nil {
    return err
}
// Background enforcement
go func() {
    ticker := time.NewTicker(1 * time.Minute)
    for range ticker.C {
        policy.EnforcePolicy(ctx)
    }
}()

type Reference added in v1.1.3

type Reference string

Reference is a reference to media stored in a backend. The format and meaning is backend-specific.

Directories

Path Synopsis
Package local provides local filesystem-based storage implementation.
Package local provides local filesystem-based storage implementation.
Package policy provides storage retention and cleanup policy management.
Package policy provides storage retention and cleanup policy management.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL