vector

package
v0.0.7 Latest Latest
Warning

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

 Go to latest Published: Jul 8, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrLeaderUnavailable = errors.New("cluster leader is unavailable")

Functions

func GetEmbedding 

func GetEmbedding(ctx context.Context, apiKey, baseUrl, modelId, text string) ([]float32, error)

GetEmbedding generates a vector embedding for the specified text using an OpenAI-compatible API.

Types

type ApplyFunc 

type ApplyFunc func(ReplicatedOperation) error

ApplyFunc applies a non-vector replicated operation on the receiving node.

type ClusterView 

type ClusterView struct {
	Mode     Mode        `json:"mode"`
	SelfAddr string      `json:"selfAddr"`
	NodeID   string      `json:"nodeId"`
	LeaderID string      `json:"leaderId"`
	IsLeader bool        `json:"isLeader"`
	Term     uint64      `json:"term"`
	Members  []PeerState `json:"members"`
}

ClusterView is a point-in-time snapshot of the cluster membership used by the monitoring panel.

type Config 

type Config struct {
	DataDsn        string
	RedisDsn       string
	DataDir        string
	NodeID         string
	Peers          []string
	EmbeddingModel string
}

Config defines the initial vector runtime configuration.

type Coordinator 

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

Coordinator implements a lightweight Raft-inspired coordination layer. It is intentionally small: leadership is derived deterministically from the set of currently-alive members (the lexicographically smallest reachable address wins) and the term is bumped whenever leadership changes. This is enough to guarantee a single writer for vector/cache state replication without pulling in a full consensus implementation.

func NewCoordinator 

func NewCoordinator(manager *Manager, config CoordinatorConfig) *Coordinator

NewCoordinator creates a coordinator for the given manager.

func (*Coordinator) AddPeer 

func (c *Coordinator) AddPeer(peer string)

AddPeer dynamically adds a new peer address to the coordinator's list.

func (*Coordinator) AdvanceIndexLocally 

func (c *Coordinator) AdvanceIndexLocally()

AdvanceIndexLocally manually increments the cluster logs indices on local save.

func (*Coordinator) AppliedLogIndex 

func (c *Coordinator) AppliedLogIndex() uint64

AppliedLogIndex returns the coordinator's applied log index.

func (*Coordinator) ApplyReplicated 

func (c *Coordinator) ApplyReplicated(op ReplicatedOperation) (Snapshot, error)

ApplyReplicated applies an operation received from the leader.

func (*Coordinator) HasPeers 

func (c *Coordinator) HasPeers() bool

HasPeers reports whether the coordinator is configured for multi-instance operation.

func (*Coordinator) InstallSnapshot 

func (c *Coordinator) InstallSnapshot(replacePath string, lastLogIndex uint64, appliedLogIndex uint64, term uint64) error

InstallSnapshot is invoked on a follower to replace its physical SQLite state machine.

func (*Coordinator) IsLeader 

func (c *Coordinator) IsLeader() bool

IsLeader reports whether this node is currently the cluster leader.

func (*Coordinator) LastLogIndex 

func (c *Coordinator) LastLogIndex() uint64

LastLogIndex returns the coordinator's last log index.

func (*Coordinator) Propose 

func (c *Coordinator) Propose(op Operation) (Snapshot, error)

Propose applies an operation through the cluster. On the leader the operation is applied locally and replicated to the followers. On a follower it is forwarded to the leader; if forwarding fails it is applied locally as a best-effort fallback so single writes never get lost.

func (*Coordinator) ProposeReplicated 

func (c *Coordinator) ProposeReplicated(op ReplicatedOperation) (Snapshot, error)

ProposeReplicated applies a common replicated operation through the cluster. Strict operations are never applied locally on a follower if leader forwarding fails; this is required for SQLite primary database consistency.

func (*Coordinator) RaftTerm 

func (c *Coordinator) RaftTerm() uint64

RaftTerm returns the coordinator's term.

func (*Coordinator) ReceiveHeartbeat 

func (c *Coordinator) ReceiveHeartbeat(hb Heartbeat) HeartbeatReply

ReceiveHeartbeat processes an inbound heartbeat from a peer.

func (*Coordinator) Start 

func (c *Coordinator) Start()

Start launches the background heartbeat/election loop.

func (*Coordinator) Stop 

func (c *Coordinator) Stop()

Stop terminates the background loop.

func (*Coordinator) Transport 

func (c *Coordinator) Transport() Transport

Transport returns the coordinator's communication transport.

func (*Coordinator) View 

func (c *Coordinator) View() ClusterView

View returns the current cluster view for monitoring.

type CoordinatorConfig 

type CoordinatorConfig struct {
	SelfAddr  string
	NodeID    string
	Peers     []string
	Interval  time.Duration
	Timeout   time.Duration
	Transport Transport
	Apply     ApplyFunc
	App       SnapshotApp
}

CoordinatorConfig configures a new coordinator.

type DBEngine 

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

DBEngine persists vector runtime state via the application's params table.

func NewDBEngine 

func NewDBEngine(dao *daos.Dao, key string) *DBEngine

NewDBEngine creates a new param-backed engine.

func (*DBEngine) Load 

func (e *DBEngine) Load() (Snapshot, bool, error)

Load restores the vector snapshot from the params table.

func (*DBEngine) Save 

func (e *DBEngine) Save(snapshot Snapshot) error

Save stores the snapshot in the params table.

type DBEntryStore 

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

DBEntryStore persists vector entries in the database.

func NewDBEntryStore 

func NewDBEntryStore(dao *daos.Dao) *DBEntryStore

NewDBEntryStore creates a database-backed entry store.

func (*DBEntryStore) Load 

func (s *DBEntryStore) Load() ([]*models.VectorEntry, error)

Load returns all persisted vector entries.

func (*DBEntryStore) Replace 

func (s *DBEntryStore) Replace(entries []*models.VectorEntry) error

Replace rewrites the vector entry table with the provided state.

type DBTaskStore 

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

DBTaskStore persists vector tasks in the database.

func NewDBTaskStore 

func NewDBTaskStore(dao *daos.Dao) *DBTaskStore

NewDBTaskStore creates a database-backed task store.

func (*DBTaskStore) Load 

func (s *DBTaskStore) Load() ([]EmbeddingTask, error)

Load returns all queued tasks ordered by creation time.

func (*DBTaskStore) Replace 

func (s *DBTaskStore) Replace(tasks []EmbeddingTask) error

Replace rewrites the task table with the provided queue state.

type EmbeddingTask 

type EmbeddingTask struct {
	Id           string    `json:"id"`
	ProjectID    string    `json:"projectId"`
	SourceType   string    `json:"sourceType"`
	SourceID     string    `json:"sourceId"`
	SourceField  string    `json:"sourceField"`
	Model        string    `json:"model"`
	ContentHash  string    `json:"contentHash"`
	Status       string    `json:"status"`
	QueuedAt     time.Time `json:"queuedAt"`
	AttemptCount int       `json:"attemptCount"`
	Payload      []byte    `json:"payload,omitempty"`
}

EmbeddingTask describes a queued embedding job.

func BuildRecordEmbeddingTask 

func BuildRecordEmbeddingTask(record *models.Record, field *schema.SchemaField, embeddingModel string) *EmbeddingTask

BuildRecordEmbeddingTask builds a queued embedding task from a record.

type Engine 

type Engine interface {
	Load() (Snapshot, bool, error)
	Save(Snapshot) error
}

Engine defines the storage backend for vector runtime state.

type EntryStore 

type EntryStore interface {
	Load() ([]*models.VectorEntry, error)
	Replace([]*models.VectorEntry) error
}

EntryStore persists the computed vector entries.

type FileEngine 

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

FileEngine is a file-backed state machine for vector snapshots.

func NewFileEngine 

func NewFileEngine(baseDir string) *FileEngine

NewFileEngine creates a file-backed engine.

func (*FileEngine) Load 

func (e *FileEngine) Load() (Snapshot, bool, error)

func (*FileEngine) Save 

func (e *FileEngine) Save(snapshot Snapshot) error

type HTTPTransport 

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

HTTPTransport implements Transport over HTTP. Peer addresses may be either a bare host:port or a full base URL. The transport posts JSON payloads to the embedded vector cluster endpoints exposed by every node.

func NewHTTPTransport 

func NewHTTPTransport(client *http.Client, token string) *HTTPTransport

NewHTTPTransport creates a new HTTP-based cluster transport. The optional token is sent as an Authorization bearer header so internal cluster traffic can be authenticated.

func (*HTTPTransport) Forward 

func (t *HTTPTransport) Forward(ctx context.Context, peer string, op ReplicatedOperation) error

Forward proxies an operation to the cluster leader.

func (*HTTPTransport) Join 

func (t *HTTPTransport) Join(ctx context.Context, leader string, selfAddr string) error

Join posts a join request to the cluster leader to dynamically register selfAddr.

func (*HTTPTransport) Replicate 

func (t *HTTPTransport) Replicate(ctx context.Context, peer string, op ReplicatedOperation) error

Replicate pushes an operation to the peer follower.

func (*HTTPTransport) SendHeartbeat 

func (t *HTTPTransport) SendHeartbeat(ctx context.Context, peer string, hb Heartbeat) (HeartbeatReply, error)

SendHeartbeat posts a heartbeat to the peer and decodes the reply.

func (*HTTPTransport) SendSnapshot 

func (t *HTTPTransport) SendSnapshot(ctx context.Context, peer string, snapshotReader io.Reader, lastLogIndex uint64, appliedLogIndex uint64, term uint64) error

SendSnapshot posts a database snapshot file to the peer follower.

type Heartbeat 

type Heartbeat struct {
	Term         uint64   `json:"term"`
	LeaderID     string   `json:"leaderId"`
	NodeID       string   `json:"nodeId"`
	Address      string   `json:"address"`
	Members      []string `json:"members"`
	SentAt       int64    `json:"sentAt"`
	LastLogIndex uint64   `json:"lastLogIndex,omitempty"`
}

Heartbeat is exchanged between cluster members to share liveness and the current leadership view.

type HeartbeatReply 

type HeartbeatReply struct {
	Term         uint64 `json:"term"`
	NodeID       string `json:"nodeId"`
	Address      string `json:"address"`
	LeaderID     string `json:"leaderId"`
	Accepted     bool   `json:"accepted"`
	LastLogIndex uint64 `json:"lastLogIndex,omitempty"`
}

HeartbeatReply is returned by a member after processing a heartbeat.

type Manager 

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

Manager provides a lightweight runtime model for the embedded vector stack.

The Raft transport and sqlite-vec integration will be added in follow-up steps. For now it provides topology detection, status reporting and lifecycle hooks for the rest of the application.

func NewManager 

func NewManager(config Config) *Manager

NewManager creates a new vector manager.

func (*Manager) ApplyOperation 

func (m *Manager) ApplyOperation(operation Operation) (Snapshot, error)

ApplyOperation replays a state transition and persists the resulting state.

func (*Manager) ApplySnapshot 

func (m *Manager) ApplySnapshot(snapshot Snapshot) error

ApplySnapshot replaces the current runtime state with the provided snapshot.

func (*Manager) AttachCoordinator 

func (m *Manager) AttachCoordinator(coordinator *Coordinator)

AttachCoordinator wires a cluster coordinator to the manager so write operations can be replicated across instances.

func (*Manager) Coordinator 

func (m *Manager) Coordinator() *Coordinator

Coordinator returns the attached cluster coordinator (may be nil).

func (*Manager) DeleteEntry 

func (m *Manager) DeleteEntry(entry *models.VectorEntry) error

DeleteEntry removes a vector entry by its composite key.

func (*Manager) DequeueEmbedding 

func (m *Manager) DequeueEmbedding() (EmbeddingTask, bool)

DequeueEmbedding removes the oldest queued embedding task.

func (*Manager) EnqueueEmbedding 

func (m *Manager) EnqueueEmbedding(task EmbeddingTask) string

EnqueueEmbedding records a new embedding task and returns its id. When a cluster coordinator is attached the enqueue is proposed through it so the task queue stays consistent across instances.

func (*Manager) Entries 

func (m *Manager) Entries() []*models.VectorEntry

Entries returns a copy of the current vector entries.

func (*Manager) IsLeader 

func (m *Manager) IsLeader() bool

IsLeader reports whether this node may execute coordinated write tasks. In standalone mode it always returns true.

func (*Manager) Load 

func (m *Manager) Load() error

Load restores the manager state from the configured snapshot file.

func (*Manager) Metrics 

func (m *Manager) Metrics() Metrics

Metrics collects the current single-node runtime metrics.

func (*Manager) Persist 

func (m *Manager) Persist() error

Persist writes the manager state to the configured snapshot file.

func (*Manager) ReplaceEntries 

func (m *Manager) ReplaceEntries(entries []*models.VectorEntry) error

ReplaceEntries replaces all vector entries and persists them.

func (*Manager) Replayable 

func (m *Manager) Replayable() Operation

Replayable converts the current status into a replay operation.

func (*Manager) SetEngine 

func (m *Manager) SetEngine(engine Engine)

SetEngine replaces the current storage backend.

func (*Manager) SetEntryStore 

func (m *Manager) SetEntryStore(store EntryStore)

SetEntryStore replaces the current vector entry persistence backend.

func (*Manager) SetTaskStore 

func (m *Manager) SetTaskStore(store TaskStore)

SetTaskStore replaces the current task persistence backend.

func (*Manager) Snapshot 

func (m *Manager) Snapshot() Snapshot

Snapshot returns the runtime status and queued embedding tasks.

func (*Manager) Start 

func (m *Manager) Start()

Start marks the vector runtime as active.

func (*Manager) Status 

func (m *Manager) Status() Status

Status returns a copy of the current vector runtime status.

func (*Manager) Stop 

func (m *Manager) Stop()

Stop marks the vector runtime as inactive.

func (*Manager) Tasks 

func (m *Manager) Tasks() []EmbeddingTask

Tasks returns a copy of the current queued embedding tasks.

func (*Manager) TriggerRecordEmbedding 

func (m *Manager) TriggerRecordEmbedding(record *models.Record) []string

TriggerRecordEmbedding queues embedding tasks for the record.

func (*Manager) UpdateCounters 

func (m *Manager) UpdateCounters(pendingEmbeddings, cacheItems int)

UpdateCounters refreshes the runtime counters used by the admin panel.

func (*Manager) UpdateEmbeddingModel 

func (m *Manager) UpdateEmbeddingModel(model string)

UpdateEmbeddingModel refreshes the embedding model tracking info.

func (*Manager) UpdateTopology 

func (m *Manager) UpdateTopology(mode Mode, leaderID string, raftTerm uint64, peers []string)

UpdateTopology refreshes the runtime status with the provided cluster hints.

func (*Manager) UpsertEntry 

func (m *Manager) UpsertEntry(entry *models.VectorEntry) error

UpsertEntry inserts or replaces a single vector entry by its composite key.

type Metrics 

type Metrics struct {
	NodeID            string    `json:"nodeId"`
	Mode              Mode      `json:"mode"`
	Online            bool      `json:"online"`
	UptimeSeconds     int64     `json:"uptimeSeconds"`
	Goroutines        int       `json:"goroutines"`
	MemAllocBytes     uint64    `json:"memAllocBytes"`
	MemSysBytes       uint64    `json:"memSysBytes"`
	HeapObjects       uint64    `json:"heapObjects"`
	GCCount           uint32    `json:"gcCount"`
	NumCPU            int       `json:"numCpu"`
	PendingEmbeddings int       `json:"pendingEmbeddings"`
	VectorEntries     int       `json:"vectorEntries"`
	CacheItems        int       `json:"cacheItems"`
	CacheBackend      string    `json:"cacheBackend"`
	EmbeddingModel    string    `json:"embeddingModel"`
	EmbeddingReady    bool      `json:"embeddingReady"`
	RedisEnabled      bool      `json:"redisEnabled"`
	Backend           string    `json:"backend"`
	DataDriver        string    `json:"dataDriver"`
	CollectedAt       time.Time `json:"collectedAt"`
}

Metrics describes the single-node runtime metrics surfaced by the monitoring panel. Cluster-wide metrics are derived from the coordinator view.

type Mode 

type Mode string

Mode describes the current vector runtime topology. 

const (
	ModeStandalone Mode = "standalone"
	ModeCluster    Mode = "cluster"
)

type Operation 

type Operation struct {
	Type              OperationType  `json:"type"`
	Snapshot          *Snapshot      `json:"snapshot,omitempty"`
	Mode              Mode           `json:"mode,omitempty"`
	LeaderID          string         `json:"leaderId,omitempty"`
	RaftTerm          uint64         `json:"raftTerm,omitempty"`
	Peers             []string       `json:"peers,omitempty"`
	EmbeddingModel    string         `json:"embeddingModel,omitempty"`
	PendingEmbeddings int            `json:"pendingEmbeddings,omitempty"`
	CacheItems        int            `json:"cacheItems,omitempty"`
	Task              *EmbeddingTask `json:"task,omitempty"`
}

Operation is a replayable vector state transition.

func UnwrapVectorOperation 

func UnwrapVectorOperation(op ReplicatedOperation) (Operation, error)

UnwrapVectorOperation decodes the legacy vector operation payload.

type OperationType 

type OperationType string

OperationType identifies a state transition that can be replayed by a replicated state machine. 

const (
	OperationTypeReplaceSnapshot OperationType = "replace_snapshot"
	OperationTypeSetTopology     OperationType = "set_topology"
	OperationTypeSetEmbedding    OperationType = "set_embedding_model"
	OperationTypeSetCounters     OperationType = "set_counters"
	OperationTypeEnqueueTask     OperationType = "enqueue_task"
	OperationTypeDequeueTask     OperationType = "dequeue_task"
)

type PeerState 

type PeerState struct {
	Address      string    `json:"address"`
	NodeID       string    `json:"nodeId"`
	Alive        bool      `json:"alive"`
	IsLeader     bool      `json:"isLeader"`
	IsSelf       bool      `json:"isSelf"`
	Term         uint64    `json:"term"`
	LastLogIndex uint64    `json:"lastLogIndex"`
	LastSeen     time.Time `json:"lastSeen,omitempty"`
}

PeerState describes the last known state of a cluster member.

type ReplicatedOperation 

type ReplicatedOperation struct {
	ID        string                  `json:"id,omitempty"`
	Kind      ReplicatedOperationKind `json:"kind"`
	Type      string                  `json:"type"`
	Strict    bool                    `json:"strict,omitempty"`
	Payload   json.RawMessage         `json:"payload,omitempty"`
	RaftTerm  uint64                  `json:"raftTerm,omitempty"`
	LogIndex  uint64                  `json:"logIndex,omitempty"`
	CreatedAt time.Time               `json:"createdAt,omitempty"`
}

ReplicatedOperation is the common envelope for cluster-replicated state transitions. The payload schema is defined by Type and Kind.

func WrapVectorOperation 

func WrapVectorOperation(op Operation) (ReplicatedOperation, error)

WrapVectorOperation converts the legacy vector operation into the common replicated operation envelope.

type ReplicatedOperationKind 

type ReplicatedOperationKind string

ReplicatedOperationKind describes the subsystem that owns a replicated operation. SQLite state operations require strict leader forwarding, while vector operations can keep their existing best-effort behavior. 

const (
	ReplicatedOperationKindVector ReplicatedOperationKind = "vector"
	ReplicatedOperationKindSQLite ReplicatedOperationKind = "sqlite"
)

type SQLiteEngine 

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

SQLiteEngine persists vector runtime state in the dedicated SQLite database.

func NewSQLiteEngine 

func NewSQLiteEngine(db *dbx.DB, key string) *SQLiteEngine

func (*SQLiteEngine) Load 

func (e *SQLiteEngine) Load() (Snapshot, bool, error)

func (*SQLiteEngine) Save 

func (e *SQLiteEngine) Save(snapshot Snapshot) error

type SQLiteEntryStore 

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

SQLiteEntryStore persists vector entries in a dedicated SQLite database.

func NewSQLiteEntryStore 

func NewSQLiteEntryStore(db *dbx.DB) *SQLiteEntryStore

func (*SQLiteEntryStore) Load 

func (s *SQLiteEntryStore) Load() ([]*models.VectorEntry, error)

func (*SQLiteEntryStore) Replace 

func (s *SQLiteEntryStore) Replace(entries []*models.VectorEntry) error

type SQLiteTaskStore 

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

SQLiteTaskStore persists vector tasks in a dedicated SQLite database.

func NewSQLiteTaskStore 

func NewSQLiteTaskStore(db *dbx.DB) *SQLiteTaskStore

func (*SQLiteTaskStore) Load 

func (s *SQLiteTaskStore) Load() ([]EmbeddingTask, error)

func (*SQLiteTaskStore) Replace 

func (s *SQLiteTaskStore) Replace(tasks []EmbeddingTask) error

type Snapshot 

type Snapshot struct {
	Status Status          `json:"status"`
	Tasks  []EmbeddingTask `json:"tasks"`
}

Snapshot describes a runtime snapshot for restore and monitoring.

type SnapshotApp 

type SnapshotApp interface {
	CreateConsistentSnapshot(targetPath string) error
	ReloadDataDBWithReplacement(replacePath string) error
}

SnapshotApp defines the app methods required for snapshot and file replacements.

type Status 

type Status struct {
	Enabled           bool      `json:"enabled"`
	Mode              Mode      `json:"mode"`
	NodeID            string    `json:"nodeId"`
	DataDriver        string    `json:"dataDriver,omitempty"`
	RedisEnabled      bool      `json:"redisEnabled"`
	Backend           string    `json:"backend"`
	EmbeddingModel    string    `json:"embeddingModel,omitempty"`
	EmbeddingReady    bool      `json:"embeddingReady"`
	StartedAt         time.Time `json:"startedAt,omitempty"`
	LastUpdatedAt     time.Time `json:"lastUpdatedAt,omitempty"`
	LeaderID          string    `json:"leaderId,omitempty"`
	RaftTerm          uint64    `json:"raftTerm"`
	AppliedLogIndex   uint64    `json:"appliedLogIndex"`
	LastLogIndex      uint64    `json:"lastLogIndex"`
	Peers             []string  `json:"peers,omitempty"`
	PendingEmbeddings int       `json:"pendingEmbeddings"`
	CacheItems        int       `json:"cacheItems"`
}

Status describes the current vector runtime state.

type TaskStore 

type TaskStore interface {
	Load() ([]EmbeddingTask, error)
	Replace([]EmbeddingTask) error
}

TaskStore persists the queued embedding tasks.

type Transport 

type Transport interface {
	SendHeartbeat(ctx context.Context, peer string, hb Heartbeat) (HeartbeatReply, error)
	Replicate(ctx context.Context, peer string, op ReplicatedOperation) error
	Forward(ctx context.Context, peer string, op ReplicatedOperation) error
	SendSnapshot(ctx context.Context, peer string, snapshotReader io.Reader, lastLogIndex uint64, appliedLogIndex uint64, term uint64) error
	Join(ctx context.Context, leader string, selfAddr string) error
}

Transport abstracts the peer-to-peer communication used by the coordinator so it can be backed by HTTP in production and an in-memory bus in tests.

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.