event

Documentation

Overview

Package event implements Event Sourcing storage

Index

• func Replay(events []Event, applyFunc func(Event) error) error
• func UnmarshalData(e Event) (any, error)
• type ArchivedData
• type ConflictDetectedData
• type CorrectedData
• type CreatedData
• type Event
  • func LoadEvents(ctx context.Context, store Loader, streamID uuid.UUID) ([]Event, error)
  • func LoadEventsAtVersion(ctx context.Context, store Loader, streamID uuid.UUID, version int) ([]Event, error)
  • func NewArchived(streamID uuid.UUID, version int, reason string) Event
  • func NewCorrected(streamID uuid.UUID, version int, correctedBy uuid.UUID, reason string) Event
  • func NewCreated(streamID uuid.UUID, data CreatedData) Event
  • func NewMerged(streamID uuid.UUID, version int, data MergedData) Event
  • func NewUpdated(streamID uuid.UUID, version int, data UpdatedData) Event
• type HistoryEntry
  • func LoadHistory(ctx context.Context, store interface{ ... }, id uuid.UUID) ([]HistoryEntry, error)
• type Loader
• type MemoryMergedData
• type MemorySource
• type MergedData
• type RelationCreatedData
• type Store
  • func NewStore(db *pgxpool.Pool, log logger.Logger) *Store
  • func (s *Store) Append(ctx context.Context, event Event) error
  • func (s *Store) AppendBatch(ctx context.Context, events []Event) error
  • func (s *Store) GetCurrentVersion(ctx context.Context, streamID uuid.UUID) (int, error)
  • func (s *Store) GetStreamsModifiedAfter(ctx context.Context, after time.Time) ([]uuid.UUID, error)
  • func (s *Store) Load(ctx context.Context, streamID uuid.UUID) ([]Event, error)
  • func (s *Store) LoadFrom(ctx context.Context, streamID uuid.UUID, fromVersion int) ([]Event, error)
  • func (s *Store) StreamExists(ctx context.Context, streamID uuid.UUID) (bool, error)
• type Type
• type UpdatedData

Functions

func Replay

func Replay(events []Event, applyFunc func(Event) error) error

Replay reconstructs memory state from a series of events. This is the core of event sourcing - state is derived from events.

func UnmarshalData

func UnmarshalData(e Event) (any, error)

UnmarshalData unmarshals event data based on event type

Types

type ArchivedData

type ArchivedData struct {
	Reason string `json:"reason"`
}

ArchivedData contains data for memory archival events

type ConflictDetectedData

type ConflictDetectedData struct {
	MemoryID1  uuid.UUID `json:"memory_id_1"`
	MemoryID2  uuid.UUID `json:"memory_id_2"`
	Similarity float32   `json:"similarity"`
	Strategy   string    `json:"strategy"` // "deactivate", "merge", "relate"
}

ConflictDetectedData contains data for conflict detection events

type CorrectedData

type CorrectedData struct {
	CorrectedBy uuid.UUID `json:"corrected_by"` // ID of the correcting memory
	Reason      string    `json:"reason"`       // Why this was corrected
}

CorrectedData contains data for memory correction events

type CreatedData

type CreatedData struct {
	SemanticID string           `json:"semantic_id"`
	Content    string           `json:"content"`
	Type       string           `json:"type"` // Memory type, will be resolved later
	Keywords   []string         `json:"keywords"`
	Entities   []extract.Entity `json:"entities"`
	Confidence float32          `json:"confidence"`
	Attributes map[string]any   `json:"attributes,omitempty"`
	Source     MemorySource     `json:"source,omitempty"`     // Who/how this memory was created
	Supersedes []uuid.UUID      `json:"supersedes,omitempty"` // Memories this replaces
}

CreatedData contains data for memory creation events

type Event

type Event struct {
	ID        uuid.UUID       `json:"id"`
	StreamID  uuid.UUID       `json:"stream_id"` // Memory ID
	Type      Type            `json:"type"`
	Version   int             `json:"version"`
	Timestamp time.Time       `json:"timestamp"`
	Data      json.RawMessage `json:"data"`
}

Event represents a domain event in the memory system

func LoadEvents

func LoadEvents(ctx context.Context, store Loader, streamID uuid.UUID) ([]Event, error)

LoadEvents loads events for a given stream ID.

func LoadEventsAtVersion

func LoadEventsAtVersion(ctx context.Context, store Loader, streamID uuid.UUID, version int) ([]Event, error)

LoadEventsAtVersion loads events up to a specific version.

func NewArchived

func NewArchived(streamID uuid.UUID, version int, reason string) Event

NewArchived creates a new memory archival event

func NewCorrected

func NewCorrected(streamID uuid.UUID, version int, correctedBy uuid.UUID, reason string) Event

NewCorrected creates a new memory correction event This marks a memory as having been corrected by another memory

func NewCreated

func NewCreated(streamID uuid.UUID, data CreatedData) Event

NewCreated creates a new memory creation event

func NewMerged

func NewMerged(streamID uuid.UUID, version int, data MergedData) Event

NewMerged creates a new memory merge event

func NewUpdated

func NewUpdated(streamID uuid.UUID, version int, data UpdatedData) Event

NewUpdated creates a new memory update event

type HistoryEntry

type HistoryEntry struct {
	Version     int       `json:"version"`
	Type        Type      `json:"type"`
	Timestamp   time.Time `json:"timestamp"`
	Description string    `json:"description"`
}

HistoryEntry represents a single event in memory history

func LoadHistory

func LoadHistory(ctx context.Context, store interface {
	Load(context.Context, uuid.UUID) ([]Event, error)
}, id uuid.UUID) ([]HistoryEntry, error)

LoadHistory loads the event history for a memory

type Loader

type Loader interface {
	Load(context.Context, uuid.UUID) ([]Event, error)
}

Loader loads all events for a stream from the store.

type MemoryMergedData

type MemoryMergedData struct {
	PrimaryID     uuid.UUID   `json:"primary_id"`
	MergedIDs     []uuid.UUID `json:"merged_ids"`
	MergedContent string      `json:"merged_content"`
	TotalAccess   int         `json:"total_access"`
}

MemoryMergedData contains data for memory merge completion events

type MemorySource

type MemorySource string

MemorySource indicates the origin and trustworthiness of a memory

const (
	SourceUserDirect     MemorySource = "user_direct"     // User directly stated this
	SourceUserConfirm    MemorySource = "user_confirm"    // User confirmed this
	SourceUserCorrection MemorySource = "user_correction" // User corrected this (highest priority)
	SourceAssistantInfer MemorySource = "assistant_infer" // Assistant inferred this (default)
)

type MergedData

type MergedData struct {
	SourceIDs      []uuid.UUID      `json:"source_ids"`
	MergedContent  string           `json:"merged_content"`
	MergedKeywords []string         `json:"merged_keywords"`
	MergedEntities []extract.Entity `json:"merged_entities"`
	Strategy       string           `json:"strategy,omitempty"`
	Reason         string           `json:"reason,omitempty"`
}

MergedData contains data for memory merge events

type RelationCreatedData

type RelationCreatedData struct {
	FromID   uuid.UUID `json:"from_id"`
	ToID     uuid.UUID `json:"to_id"`
	Type     string    `json:"type"`
	Strength float32   `json:"strength"`
}

RelationCreatedData contains data for relation creation events

type Store

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

Store handles event persistence for the memory system

func NewStore

func NewStore(db *pgxpool.Pool, log logger.Logger) *Store

NewStore creates a new event store

func (*Store) Append

func (s *Store) Append(ctx context.Context, event Event) error

Append adds a new event to the event store

func (*Store) AppendBatch

func (s *Store) AppendBatch(ctx context.Context, events []Event) error

AppendBatch adds multiple events atomically to the event store All events are stored in a single transaction to ensure consistency

func (*Store) GetCurrentVersion

func (s *Store) GetCurrentVersion(ctx context.Context, streamID uuid.UUID) (int, error)

GetCurrentVersion returns the latest version for a stream

func (*Store) GetStreamsModifiedAfter

func (s *Store) GetStreamsModifiedAfter(ctx context.Context, after time.Time) ([]uuid.UUID, error)

GetStreamsModifiedAfter returns stream IDs modified after the given time

func (*Store) Load

func (s *Store) Load(ctx context.Context, streamID uuid.UUID) ([]Event, error)

Load retrieves all events for a stream

func (*Store) LoadFrom

func (s *Store) LoadFrom(ctx context.Context, streamID uuid.UUID, fromVersion int) ([]Event, error)

LoadFrom retrieves events from a specific version

func (*Store) StreamExists

func (s *Store) StreamExists(ctx context.Context, streamID uuid.UUID) (bool, error)

StreamExists checks if a stream has any events

type Type

type Type string

Type represents the type of memory event

const (
	Created  Type = "created"
	Updated  Type = "updated"
	Archived Type = "archived"
	Merged   Type = "merged"

	// Curation events
	ConflictDetected Type = "conflict_detected"
	MemoryMerged     Type = "memory_merged"
	RelationCreated  Type = "relation_created"

	// Correction event - when a memory is marked as corrected
	Corrected Type = "corrected"
)

type UpdatedData

type UpdatedData struct {
	Content    *string          `json:"content,omitempty"`
	Keywords   []string         `json:"keywords,omitempty"`
	Entities   []extract.Entity `json:"entities,omitempty"`
	Confidence *float32         `json:"confidence,omitempty"`
	Attributes map[string]any   `json:"attributes,omitempty"`
	Reason     string           `json:"reason"`
	Source     MemorySource     `json:"source,omitempty"` // Source of the update
}

UpdatedData contains data for memory update events