storage

package
v1.0.15 Latest Latest
Warning

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

 Go to latest Published: Aug 8, 2026 License: MPL-2.0 Imports: 18 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrPlatformReadUnsupported = errors.New("platform message store is write-only; query the control plane instead")

ErrPlatformReadUnsupported is returned by the read side of PlatformStore. The control plane owns queries over stored messages; an engine that could read them back would need query credentials it deliberately does not have.

Functions

This section is empty.

Types

type CompositeStore 

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

func NewCompositeStore 

func NewCompositeStore(inner MessageStore, mode string, stages []string) *CompositeStore

func (*CompositeStore) Count added in v1.0.4 

func (c *CompositeStore) Count(opts QueryOpts) (int64, error)

func (*CompositeStore) Delete 

func (c *CompositeStore) Delete(id string) error

func (*CompositeStore) Get 

func (c *CompositeStore) Get(id string) (*MessageRecord, error)

func (*CompositeStore) GetStage 

func (c *CompositeStore) GetStage(id, stage string) (*MessageRecord, error)

func (*CompositeStore) Mode 

func (c *CompositeStore) Mode() string

func (*CompositeStore) Prune 

func (c *CompositeStore) Prune(before time.Time, channel string) (int, error)

func (*CompositeStore) Query 

func (c *CompositeStore) Query(opts QueryOpts) ([]*MessageRecord, error)

func (*CompositeStore) Save 

func (c *CompositeStore) Save(record *MessageRecord) error

func (*CompositeStore) ShouldStore 

func (c *CompositeStore) ShouldStore(stage string) bool

type MemoryStore 

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

func NewMemoryStore 

func NewMemoryStore(maxRecords, maxBytes int) *MemoryStore

func (*MemoryStore) BytesUsed added in v1.0.4 

func (m *MemoryStore) BytesUsed() int

func (*MemoryStore) Count added in v1.0.4 

func (m *MemoryStore) Count(opts QueryOpts) (int64, error)

func (*MemoryStore) Delete 

func (m *MemoryStore) Delete(id string) error

func (*MemoryStore) Get 

func (m *MemoryStore) Get(id string) (*MessageRecord, error)

func (*MemoryStore) GetStage 

func (m *MemoryStore) GetStage(id, stage string) (*MessageRecord, error)

func (*MemoryStore) Len added in v1.0.4 

func (m *MemoryStore) Len() int

func (*MemoryStore) Prune 

func (m *MemoryStore) Prune(before time.Time, channel string) (int, error)

func (*MemoryStore) Query 

func (m *MemoryStore) Query(opts QueryOpts) ([]*MessageRecord, error)

func (*MemoryStore) Save 

func (m *MemoryStore) Save(record *MessageRecord) error

type MessageRecord 

type MessageRecord struct {
	ID            string
	CorrelationID string
	ChannelID     string
	Stage         string
	Content       []byte
	ContentSize   int `json:"ContentSize,omitempty"`
	Status        string
	Timestamp     time.Time
	DurationMs    int64 `json:"DurationMs,omitempty"`
	Metadata      map[string]any
}

type MessageStore 

type MessageStore interface {
	Save(record *MessageRecord) error
	Get(id string) (*MessageRecord, error)
	GetStage(id, stage string) (*MessageRecord, error)
	Query(opts QueryOpts) ([]*MessageRecord, error)
	Count(opts QueryOpts) (int64, error)
	Delete(id string) error
	Prune(before time.Time, channel string) (int, error)
}

func NewMessageStore 

func NewMessageStore(cfg *config.MessageStorageConfig) (MessageStore, error)

type PlatformStore added in v1.0.15 

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

PlatformStore ships message and stage records to a managed control plane instead of writing them locally.

It exists because a cloud deployment's operator surface is the platform's Messages and Monitor views, not the engine's own dashboard, and because a runner task must not hold database credentials: it executes customer TypeScript, so anything it can reach is reachable by that code. Posting to an authenticated ingest endpoint keeps the blast radius of a hostile channel script down to its own deployment's records.

Writes are batched and asynchronous. Message throughput must not be coupled to the control plane's latency or availability -- a channel that stops accepting HL7 because a reporting endpoint is slow has failed at its actual job. The trade-off is explicit: records are best-effort and a hard kill loses whatever is still buffered. Delivery guarantees for clinical data are the destination's business, not the reporting path's.

Only Save is supported. The read methods exist to satisfy MessageStore and return ErrPlatformReadUnsupported; their callers are the engine dashboard and the reprocess CLI, neither of which runs in a cloud task.

func NewPlatformStore added in v1.0.15 

func NewPlatformStore(cfg *config.StoragePlatformConfig, logger *slog.Logger) (*PlatformStore, error)

NewPlatformStore constructs the store and starts its flush loop.

func (*PlatformStore) Close added in v1.0.15 

func (s *PlatformStore) Close() error

Close flushes what is buffered and stops the loop. The runner calls it after the engine has drained, so the last messages of a deployment's life are not the ones that go missing.

func (*PlatformStore) Count added in v1.0.15 

func (s *PlatformStore) Count(QueryOpts) (int64, error)

func (*PlatformStore) Delete added in v1.0.15 

func (s *PlatformStore) Delete(string) error

func (*PlatformStore) Get added in v1.0.15 

func (s *PlatformStore) Get(string) (*MessageRecord, error)

func (*PlatformStore) GetStage added in v1.0.15 

func (s *PlatformStore) GetStage(string, string) (*MessageRecord, error)

func (*PlatformStore) Prune added in v1.0.15 

func (s *PlatformStore) Prune(time.Time, string) (int, error)

Prune is a no-op rather than an error: retention on the managed store is the control plane's policy (it owns the payloads and the legal-hold rules), and an engine that returned an error here would make a scheduled prune look like a failure every time it ran.

func (*PlatformStore) Query added in v1.0.15 

func (s *PlatformStore) Query(QueryOpts) ([]*MessageRecord, error)

func (*PlatformStore) Save added in v1.0.15 

func (s *PlatformStore) Save(record *MessageRecord) error

Save queues a record. It never blocks on the network.

type PostgresStore 

type PostgresStore = SQLStore

PostgresStore is a backward-compatible alias for SQLStore configured with the PostgreSQL dialect.

func NewPostgresStore 

func NewPostgresStore(cfg *config.StoragePostgresConfig) (*PostgresStore, error)

NewPostgresStore creates a PostgreSQL-backed message store using the pgx driver. This is a convenience wrapper around NewSQLStore.

type QueryOpts 

type QueryOpts struct {
	ChannelID      string
	Status         string
	Stage          string
	Since          time.Time
	Before         time.Time
	Limit          int
	Offset         int
	ExcludeContent bool
}

type S3Store 

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

func NewS3Store 

func NewS3Store(cfg *config.StorageS3Config) (*S3Store, error)

func (*S3Store) Close 

func (s *S3Store) Close() error

func (*S3Store) Count added in v1.0.4 

func (s *S3Store) Count(opts QueryOpts) (int64, error)

func (*S3Store) Delete 

func (s *S3Store) Delete(id string) error

func (*S3Store) Get 

func (s *S3Store) Get(id string) (*MessageRecord, error)

func (*S3Store) GetStage 

func (s *S3Store) GetStage(id, stage string) (*MessageRecord, error)

func (*S3Store) Prune 

func (s *S3Store) Prune(before time.Time, channel string) (int, error)

func (*S3Store) Query 

func (s *S3Store) Query(opts QueryOpts) ([]*MessageRecord, error)

func (*S3Store) Save 

func (s *S3Store) Save(record *MessageRecord) error

type SQLStore added in v1.0.15 

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

SQLStore is a database-backed MessageStore that works with any SQL backend supported by the dbutil.Dialect abstraction.

func NewSQLStore added in v1.0.15 

func NewSQLStore(driver, dsn, tablePrefix string, maxOpen, maxIdle int) (*SQLStore, error)

NewSQLStore opens a connection using the given config-level driver name and DSN, then ensures the messages table exists.

func (*SQLStore) Close added in v1.0.15 

func (s *SQLStore) Close() error

func (*SQLStore) Count added in v1.0.15 

func (s *SQLStore) Count(opts QueryOpts) (int64, error)

func (*SQLStore) Delete added in v1.0.15 

func (s *SQLStore) Delete(id string) error

func (*SQLStore) Get added in v1.0.15 

func (s *SQLStore) Get(id string) (*MessageRecord, error)

func (*SQLStore) GetStage added in v1.0.15 

func (s *SQLStore) GetStage(id, stage string) (*MessageRecord, error)

func (*SQLStore) Prune added in v1.0.15 

func (s *SQLStore) Prune(before time.Time, channel string) (int, error)

func (*SQLStore) Query added in v1.0.15 

func (s *SQLStore) Query(opts QueryOpts) ([]*MessageRecord, error)

func (*SQLStore) Save added in v1.0.15 

func (s *SQLStore) Save(record *MessageRecord) error

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.