nodra

package module

Go to main page

Versions in this module

v1

v1.1.0

Oct 25, 2026

v1.0.0

Oct 25, 2026

v0

v0.13.0

Oct 25, 2026

v0.12.1

Oct 25, 2026

v0.12.0

Oct 25, 2026

v0.11.1

Oct 25, 2026

v0.11.0

Oct 25, 2026

v0.9.1

Oct 25, 2026

Changes in this version

type Store

+ func (s *Store[T]) BatchErr(fn func() error) error

v0.9.0

Oct 25, 2026

Changes in this version

+ var ErrInvalidKey = errors.New("nodra: invalid encryption key")

+ var ErrPersistenceRecovery = errors.New("nodra: primary and backup recovery failed")

+ func Atomic2[A, B any](first *Store[A], second *Store[B], fn func(*A, *B) error) (err error)

+ func AtomicGroup(ops ...AtomicOp) (err error)

+ func InvalidateTarget[T any](source Observable[T], target InvalidationTarget) func()

+ type AESGCMCodec struct

+ Inner Codec[T]

+ Keys KeyProvider

+ func (c AESGCMCodec[T]) Decode(data []byte) (T, error)

+ func (c AESGCMCodec[T]) Encode(value T) ([]byte, error)

+ type ActionMetrics struct

+ Accepted uint64

+ Canceled uint64

+ Completed uint64

+ DurationMax time.Duration

+ DurationSum time.Duration

+ Failed uint64

+ QueuePeak int

+ Retried uint64

+ Succeeded uint64

type ActionOptions

+ Retry RetryOptions

type AsyncAction

+ func (a *AsyncAction[T, I, R]) ExportMetrics(sink MetricsSink) error

+ func (a *AsyncAction[T, I, R]) Metrics() ActionMetrics

+ func (a *AsyncAction[T, I, R]) SubscribeMetrics(fn func(ActionMetrics)) func()

+ type AtomicOp struct

+ func AtomicUpdate[T any](store *Store[T], fn func(*T) error) AtomicOp

type FileStorage

+ Durable bool

+ Journal bool

+ type InvalidationTarget interface

+ Invalidate func()

+ type KeyProvider interface

+ Current func() (id string, key []byte, err error)

+ Get func(id string) ([]byte, error)

+ type MemoryKeyProvider struct

+ func NewMemoryKeyProvider(id string, key []byte) (*MemoryKeyProvider, error)

+ func (p *MemoryKeyProvider) Add(id string, key []byte) error

+ func (p *MemoryKeyProvider) Current() (string, []byte, error)

+ func (p *MemoryKeyProvider) Get(id string) ([]byte, error)

+ func (p *MemoryKeyProvider) SetCurrent(id string) error

+ type MetricsSink interface

+ RecordActionMetrics func(name string, metrics ActionMetrics)

type Mutation

+ TransactionID uint64

type MutationType

+ const MutationTransaction

type Persistence

+ func (p *Persistence[T]) LoadWithRecovery(ctx context.Context) error

+ type RecoveryError struct

+ Backup error

+ Primary error

+ func (e *RecoveryError) Error() string

+ func (e *RecoveryError) Unwrap() []error

+ type RetryOptions struct

+ Backoff time.Duration

+ Idempotent bool

+ MaxAttempts int

+ MaxBackoff time.Duration

type Snapshot

+ TransactionID uint64

type TaskMetrics

+ Attempts int

v0.7.0

Oct 24, 2026

Changes in this version

+ var ErrFileConflict = errors.New("nodra: file content changed since last observation")

+ var ErrInvalidBinding = errors.New("nodra: invalid field binding")

+ var ErrInvalidHistory = errors.New("nodra: invalid history size configuration")

type ActionEvent

+ Metrics TaskMetrics

type ActionOptions

+ Debounce time.Duration

type ActionTrace

+ Metrics TaskMetrics

type AsyncPersistence

+ func (a *AsyncPersistence[T]) Status() ReadOnly[PersistenceState]

type AsyncState

+ Metrics TaskMetrics

+ type CheckedFileStorage struct

+ func (c *CheckedFileStorage) Load(ctx context.Context) (data []byte, err error)

+ func (c *CheckedFileStorage) LoadBackup(ctx context.Context) (data []byte, err error)

+ func (c *CheckedFileStorage) Save(ctx context.Context, data []byte) error

type Derived

+ func Combine[A, B, V any](first Observable[A], second Observable[B], project func(A, B) V, ...) *Derived[Pair[A, B], V]

+ func (d *Derived[T, V]) Snapshot() Snapshot[V]

+ func (d *Derived[T, V]) SubscribeSnapshot(fn func(Snapshot[V])) func()

+ type Field struct

+ func BindField[T, V any](store *Store[T], read func(T) V, write func(*T, V) error) (Field[T, V], error)

+ func (f Field[T, V]) Edit(edit func(*V) bool) error

+ func (f Field[T, V]) Get() V

+ func (f Field[T, V]) Set(value V) error

+ type FileConflictError struct

+ Actual FileFingerprint

+ Expected FileFingerprint

+ Path string

+ func (e *FileConflictError) Error() string

+ func (e *FileConflictError) Unwrap() error

+ type FileFingerprint struct

+ Exists bool

+ SHA256 [sha256.Size]byte

type FileStorage

+ func (f FileStorage) WithConflictDetection(ctx context.Context) (*CheckedFileStorage, error)

type History

+ func (h *History[T]) Status() ReadOnly[HistoryState]

type HistoryOptions

+ MaxBytes int64

+ Size func(T) int64

+ type HistoryState struct

+ Bytes int64

+ CanRedo bool

+ CanUndo bool

+ Closed bool

+ Dropped uint64

+ Grouped bool

+ MaxBytes int64

+ RedoEntries int

+ Revision uint64

+ UndoEntries int

+ type Observable interface

+ Snapshot func() Snapshot[T]

+ SubscribeSnapshot func(func(Snapshot[T])) func()

+ type Pair struct

+ First A

+ Second B

+ type PersistenceState struct

+ Closed bool

+ Error string

+ HasSaved bool

+ Pending bool

+ RequestedVersion uint64

+ Revision uint64

+ SavedVersion uint64

+ Saving bool

type ReadOnly

+ func (r ReadOnly[T]) SubscribeSnapshot(fn func(Snapshot[T])) func()

type Reader

+ SubscribeSnapshot func(func(Snapshot[T])) func()

type Store

+ func (s *Store[T]) SubscribeSnapshot(fn func(Snapshot[T])) func()

+ type TaskMetrics struct

+ ActiveWorkers int

+ Commit time.Duration

+ Debounce time.Duration

+ Queue time.Duration

+ QueueDepth int

+ Work time.Duration

v0.6.0

Oct 24, 2026

Changes in this version

+ var ErrHistoryGroup = errors.New("nodra: cannot undo or redo inside an open history group")

+ var ErrInvalidAction = errors.New("nodra: invalid async action")

+ var ErrNoBackup = errors.New("nodra: backup storage unavailable")

+ var ErrQueueFull = errors.New("nodra: async action queue full")

+ var ErrSuperseded = errors.New("nodra: async action superseded")

+ func InvalidateProjectionWith[T, V any](source Reader[T], window *mygo.Window, project func(T) V, ...) func()

+ func InvalidateProjection[T, V any](source Reader[T], window *mygo.Window, project func(T) V) func()

+ func SubscribeProjectionWith[T, V any](source Reader[T], project func(T) V, fn func(current, previous V), ...) func()

+ func SubscribeProjection[T, V any](source Reader[T], project func(T) V, fn func(V)) func()

type ActionEvent

+ Committed bool

+ ID uint64

+ Key string

+ RunID uint64

+ Version uint64

+ type ActionOptions struct

+ Key func(I) string

+ MaxConcurrent int

+ Policy ActionPolicy

+ QueueLimit int

+ type ActionPolicy uint8

+ const LatestWins

+ const QueueAll

+ type ActionTrace struct

+ At time.Time

+ Committed bool

+ Duration time.Duration

+ Err error

+ ID uint64

+ Key string

+ Name string

+ Phase TaskPhase

+ Progress float64

+ Revision uint64

+ RunID uint64

+ StoreID string

+ TaskVersion uint64

+ Version uint64

+ type AsyncAction struct

+ func NewAsyncActionWithOptions[T, I, R any](store *Store[T], name string, ...) (*AsyncAction[T, I, R], error)

+ func NewAsyncAction[T, I, R any](store *Store[T], name string, ...) (*AsyncAction[T, I, R], error)

+ func (a *AsyncAction[T, I, R]) Close(ctx context.Context) error

+ func (a *AsyncAction[T, I, R]) Run(ctx context.Context, input I) (*Task, error)

+ func (a *AsyncAction[T, I, R]) Status() ReadOnly[AsyncState]

+ type AsyncState struct

+ Error string

+ ID uint64

+ Key string

+ Name string

+ Phase TaskPhase

+ Progress float64

+ Revision uint64

+ Running bool

type Derived

+ func DeriveProjection[T any, V any](source Reader[T], project func(T) V, equals func(V, V) bool) *Derived[T, V]

type FileStorage

+ Backup bool

+ func (f FileStorage) LoadBackup(ctx context.Context) ([]byte, error)

type History

+ func TrackWith[T any](store *Store[T], options HistoryOptions[T]) *History[T]

+ func (h *History[T]) BeginGroup() (func(), error)

+ type HistoryOptions struct

+ Equals func(previous, next T) bool

+ Limit int

+ Restore func(current, recorded T) T

+ type LoadError struct

+ Backup bool

+ Err error

+ Stage string

+ func (e *LoadError) Error() string

+ func (e *LoadError) Unwrap() error

type Mutation

+ ActionID uint64

+ type PanicError struct

+ Value any

+ func (e *PanicError) Error() string

type Persistence

+ func NewPersistenceWithOptions[T any](store *Store[T], storage Storage, codec Codec[T], ...) *Persistence[T]

+ func (p *Persistence[T]) LoadBackup(ctx context.Context) error

+ type PersistenceOptions struct

+ Validate func(T) error

+ type ReadOnly struct

+ func (r ReadOnly[T]) ActiveTasks() []ActionTrace

+ func (r ReadOnly[T]) Done() <-chan struct{}

+ func (r ReadOnly[T]) Get() T

+ func (r ReadOnly[T]) ID() string

+ func (r ReadOnly[T]) Snapshot() Snapshot[T]

+ func (r ReadOnly[T]) State() T

+ func (r ReadOnly[T]) Subscribe(fn func(Snapshot[T])) func()

+ func (r ReadOnly[T]) SubscribeActionTrace(fn func(ActionTrace)) func()

+ func (r ReadOnly[T]) SubscribeActions(fn func(ActionEvent)) func()

+ func (r ReadOnly[T]) SubscribeMutations(fn func(Mutation[T])) func()

+ func (r ReadOnly[T]) SubscribeWith(options SubscribeOptions[T], fn func(Snapshot[T])) func()

+ func (r ReadOnly[T]) TaskSnapshot() TaskSnapshot

+ func (r ReadOnly[T]) Version() uint64

+ func (r ReadOnly[T]) Watch(ctx context.Context, buffer int) <-chan Snapshot[T]

+ type Reader interface

+ ActiveTasks func() []ActionTrace

+ Done func() <-chan struct{}

+ Get func() T

+ ID func() string

+ Snapshot func() Snapshot[T]

+ State func() T

+ Subscribe func(func(Snapshot[T])) func()

+ SubscribeActionTrace func(func(ActionTrace)) func()

+ SubscribeActions func(func(ActionEvent)) func()

+ SubscribeMutations func(func(Mutation[T])) func()

+ SubscribeWith func(SubscribeOptions[T], func(Snapshot[T])) func()

+ TaskSnapshot func() TaskSnapshot

+ Version func() uint64

+ Watch func(context.Context, int) <-chan Snapshot[T]

+ type Scope struct

+ func NewScope() *Scope

+ func (s *Scope) Close(ctx context.Context) error

+ func (s *Scope) Context() context.Context

+ func (s *Scope) Defer(cleanup func(context.Context) error) error

+ func (s *Scope) Done() <-chan struct{}

+ func (s *Scope) Observe(stop func()) error

type Snapshot

+ func Select[T, V any](source Reader[T], project func(T) V) Snapshot[V]

type Store

+ func (s *Store[T]) ActiveTasks() []ActionTrace

+ func (s *Store[T]) Done() <-chan struct{}

+ func (s *Store[T]) ReadOnly() ReadOnly[T]

+ func (s *Store[T]) SubscribeActionTrace(fn func(ActionTrace)) func()

+ func (s *Store[T]) TaskSnapshot() TaskSnapshot

+ type Task struct

+ ID uint64

+ func (t *Task) Cancel()

+ func (t *Task) Done() <-chan struct{}

+ func (t *Task) Status() ReadOnly[AsyncState]

+ func (t *Task) Wait(ctx context.Context) error

+ type TaskPhase string

+ const TaskCanceled

+ const TaskFailed

+ const TaskQueued

+ const TaskRunning

+ const TaskSucceeded

+ const TaskSuperseded

+ type TaskSnapshot struct

+ Revision uint64

+ Tasks []ActionTrace

v0.5.0

Oct 24, 2026

Changes in this version

+ func InvalidateSelectorWith[T any, V any](store *Store[T], window *mygo.Window, selector func(T) V, ...) (stop func())

+ func InvalidateSelector[T any, V any](store *Store[T], window *mygo.Window, selector func(T) V) (stop func())

v0.4.0

Oct 24, 2026

v0.3.0

Oct 24, 2026

Changes in this version

+ var ErrHistoryBatch = errors.New("nodra: cannot undo or redo inside an open batch")

+ var ErrMissingMigration = errors.New("nodra: missing schema migration")

+ var ErrUnsupportedVersion = errors.New("nodra: unsupported schema version")

+ type AsyncOptions struct

+ Debounce time.Duration

+ type AsyncPersistence struct

+ func (a *AsyncPersistence[T]) Close(ctx context.Context) error

+ func (a *AsyncPersistence[T]) Errors() <-chan error

+ func (a *AsyncPersistence[T]) Flush(ctx context.Context) error

type Definition

+ func DefineWithClone[T any](id string, initial func() T, clone func(T) T) *Definition[T]

+ type JSONMigration func(json.RawMessage) (json.RawMessage, error)

type Persistence

+ func (p *Persistence[T]) AutoSaveAsync(ctx context.Context, options AsyncOptions) (*AsyncPersistence[T], error)

type Store

+ func NewWithClone[T any](initial T, clone func(T) T) *Store[T]

+ type VersionedJSONCodec struct

+ AllowUnversioned bool

+ Migrations map[int]JSONMigration

+ Version int

+ func (c VersionedJSONCodec[T]) Decode(data []byte) (T, error)

+ func (c VersionedJSONCodec[T]) Encode(value T) ([]byte, error)

v0.2.0

Oct 21, 2026

Changes in this version

+ func Invalidate[T any](store *Store[T], window *mygo.Window) (stop func())

v0.1.0

Oct 21, 2026

Changes in this version

+ var ErrClosed = errors.New("nodra: store is closed")

+ var ErrInvalidPersistence = errors.New("nodra: invalid persistence configuration")

+ var ErrNoRedo = errors.New("nodra: nothing to redo")

+ var ErrNoUndo = errors.New("nodra: nothing to undo")

+ func SubscribeSelectorWith[T any, V comparable](store *Store[T], selector func(T) V, fn func(current, previous V), ...) func()

+ func SubscribeSelector[T any, S comparable](store *Store[T], selector func(T) S, fn func(S)) func()

+ type ActionEvent struct

+ Duration time.Duration

+ Err error

+ Finished time.Time

+ Name string

+ Started time.Time

+ StoreID string

+ type ActionHooks struct

+ After func(ActionEvent)

+ Before func(ActionEvent)

+ Error func(ActionEvent)

+ type Codec interface

+ Decode func([]byte) (T, error)

+ Encode func(T) ([]byte, error)

+ type Definition struct

+ func Define[T any](id string, initial func() T) *Definition[T]

+ func (d *Definition[T]) ID() string

+ func (d *Definition[T]) New() *Store[T]

+ type Derived struct

+ func Derive[T any, V any](store *Store[T], selector func(T) V, equals func(V, V) bool) *Derived[T, V]

+ func (d *Derived[T, V]) Close() error

+ func (d *Derived[T, V]) Get() V

+ func (d *Derived[T, V]) Subscribe(fn func(V)) func()

+ type FileStorage struct

+ Path string

+ Perm os.FileMode

+ func (f FileStorage) Load(ctx context.Context) ([]byte, error)

+ func (f FileStorage) Save(ctx context.Context, data []byte) (err error)

+ type History struct

+ func Track[T any](store *Store[T], limit int) *History[T]

+ func (h *History[T]) CanRedo() bool

+ func (h *History[T]) CanUndo() bool

+ func (h *History[T]) Close() error

+ func (h *History[T]) Redo() error

+ func (h *History[T]) Undo() error

+ type JSONCodec struct

+ func (JSONCodec[T]) Decode(data []byte) (T, error)

+ func (JSONCodec[T]) Encode(value T) ([]byte, error)

+ type Mutation struct

+ At time.Time

+ Name string

+ Previous T

+ Type MutationType

+ Value T

+ Version uint64

+ type MutationType string

+ const MutationAction

+ const MutationDirect

+ const MutationPatch

+ const MutationReset

+ type Persistence struct

+ func NewPersistence[T any](store *Store[T], storage Storage, codec Codec[T]) *Persistence[T]

+ func (p *Persistence[T]) AutoSave(ctx context.Context) func()

+ func (p *Persistence[T]) AutoSaveWithErrors(ctx context.Context) (<-chan error, func())

+ func (p *Persistence[T]) Load(ctx context.Context) error

+ func (p *Persistence[T]) Save(ctx context.Context) error

+ type SelectorOptions struct

+ Equals func(previous, next V) bool

+ Immediate bool

+ type Snapshot struct

+ Value T

+ Version uint64

+ type Storage interface

+ Load func(context.Context) ([]byte, error)

+ Save func(context.Context, []byte) error

+ type Store struct

+ func Create[T any](initial T) *Store[T]

+ func New[T any](initial T) *Store[T]

+ func (s *Store[T]) Action(name string, fn func(*T) error) func() error

+ func (s *Store[T]) Batch(fn func()) error

+ func (s *Store[T]) Close() error

+ func (s *Store[T]) Do(name string, fn func(*T) error) error

+ func (s *Store[T]) Get() T

+ func (s *Store[T]) ID() string

+ func (s *Store[T]) Patch(fn func(*T)) error

+ func (s *Store[T]) Reset() error

+ func (s *Store[T]) Set(value T) error

+ func (s *Store[T]) Snapshot() Snapshot[T]

+ func (s *Store[T]) State() T

+ func (s *Store[T]) Subscribe(fn func(Snapshot[T])) func()

+ func (s *Store[T]) SubscribeActionHooks(hooks ActionHooks) func()

+ func (s *Store[T]) SubscribeActions(fn func(ActionEvent)) func()

+ func (s *Store[T]) SubscribeMutations(fn func(Mutation[T])) func()

+ func (s *Store[T]) SubscribeWith(options SubscribeOptions[T], fn func(Snapshot[T])) func()

+ func (s *Store[T]) Transaction(fn func(*T) error) error

+ func (s *Store[T]) Update(fn func(*T)) error

+ func (s *Store[T]) UpdateErr(fn func(*T) error) error

+ func (s *Store[T]) Version() uint64

+ func (s *Store[T]) Watch(ctx context.Context, buffer int) <-chan Snapshot[T]

+ type SubscribeOptions struct

+ Equals func(previous, next T) bool

+ Immediate bool

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