storage

package
v0.11.0 Latest Latest
Warning

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

 Go to latest Published: Aug 17, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Overview

Package storage provides dialect-aware repository interfaces and a single implementation (Store) that serves both the SQLite and PostgreSQL backends. The active dialect is chosen at construction; the repository code below is backend-agnostic (it goes through the central exec shim and the Dialect interface — see storage.go and dialect.go).

Repository Pattern:

  • All repositories are interfaces for testability
  • Store implements all 12+ interfaces
  • Methods return wrapped errors with context

Critical Invariants (MUST READ):

  • Message IDs are GLOBAL auto-increment, NOT per-user
  • ANY query using ID ranges MUST include user_id filter
  • Example: WHERE id >= ? AND id <= ? AND user_id = ?

Thread Safety (SQLite backend):

  • The SQLite store uses a single connection with WAL mode
  • Concurrent reads are safe; writes are serialized by SQLite
  • PostgreSQL uses a normal connection pool

Usage Example: 

mockStorage := testutil.NewMockStorage()
mockStorage.On("GetFacts", userID).Return(testutil.TestFacts(), nil)
svc := NewService(mockStorage, ...)

Index

Constants

View Source 
const (
	CircleFamily    = "Family"
	CircleFriends   = "Friends"
	CircleWorkInner = "Work_Inner"
	CircleWorkOuter = "Work_Outer"
	CircleOther     = "Other"
)

Circle classifies a person's relationship to the memory scope.

In a DM scope the circle is the person's relationship to the user (the bot owner). In a channel scope it marks insider vs outsider relative to the channel's participants: Work_Inner is a channel member (someone who posts in the channel), Other is an external person the participants merely mention.

View Source 
const (
	// FactKindSelfReport is the default: something the user stated about
	// themselves or their own life.
	FactKindSelfReport = "self_report"
	// FactKindUserOpinion is the user's judgment or interpretation of other
	// people and conflicts, known only from the user's side.
	FactKindUserOpinion = "user_opinion"
	// FactKindVerified is reserved for facts confirmed by evidence beyond the
	// user's account (documents, external records). Not assignable by agents.
	FactKindVerified = "verified"
	// FactKindConstraint is a user instruction that restricts the assistant's
	// behavior ("don't be harsh", "no psychological interpretations"). Stored
	// separately so the system prompt can treat it as a preference that may be
	// overridden in safety-relevant situations, not as an absolute rule.
	FactKindConstraint = "constraint"
)

FactKind records the provenance of a stored fact — how much epistemic weight it carries when re-injected into LLM context.

The distinction exists because facts are fed back into every conversation as ground truth: a user's one-sided verdict about another person ("X is manipulative") stored as a plain fact becomes something the model cannot argue with. Tagging provenance lets the context layer present such facts as testimony rather than truth.

View Source 
const (
	TagInnerCircle         = "inner_circle"         // Work_Inner + Family, system prompt
	TagRelevantPeople      = "relevant_people"      // Reranker selected, user prompt
	TagPeople              = "people"               // All people, for Archivist
	TagChannelParticipants = "channel_participants" // Active channel members (Phase 6)
)

XML tag constants for people formatting.

Variables

This section is empty.

Functions

func ComposePersonEmbeddingText added in v0.10.2 

func ComposePersonEmbeddingText(displayName string, username *string, aliases []string, bio string) string

ComposePersonEmbeddingText is the single source of truth for the text that represents a person to the embedding model: display name + username + aliases + bio. Every site that embeds a person — the live add/update/merge paths and the startup re-embed — MUST go through this function. When two sites composed it differently (the re-embed once dropped username), stored and freshly-written vectors landed in subtly different spaces and people RAG quality silently degraded for anyone with a username.

func EmbeddingVersion added in v0.10.3 

func EmbeddingVersion(model string, dim int) string

EmbeddingVersion composes the version tag written into embedding_version columns. Format: "{model}:{dim}"; a dim of 0 means "provider default" and yields just the model string. This is the single source of truth — the startup re-embed migration and the live write paths must agree on it.

func ExpandIn added in v0.9.0 

func ExpandIn(query string, args ...any) (string, []any, error)

ExpandIn rewrites a SQL query by expanding a single `?` placeholder whose matching argument is a slice into `?,?,?,...` (one `?` per slice element), and returns the rewritten query plus a flat argument list.

Rules:

  • Scalar args are passed through unchanged.
  • Exactly one slice argument is supported. Zero slices is fine (the query is returned unchanged with a copied args slice). Two or more slices return an error.
  • An empty or nil slice returns an error — `IN ()` is not valid SQL, and callers should guard the no-IDs case themselves (the early-return is usually what they want anyway).
  • Placeholder counting is naive: `?` inside SQL string literals is not skipped. SQLite does not treat `?` inside `'...'` specially, but if a query contains a literal `?` inside a string, behaviour is undefined.

Usage: 

q, args, err := ExpandIn(
    "SELECT ... FROM t WHERE user_id = ? AND id IN (?)",
    userID, ids,
)
if err != nil {
    return nil, err
}
rows, err := db.Query(q, args...)

func FormatChannelProfile added in v0.10.0 

func FormatChannelProfile(facts []Fact) string

FormatChannelProfile is the channel-scope counterpart of FormatUserProfile: the same fact rendering wrapped in <channel_profile> tags (Phase 6). For a channel scope these facts describe the channel itself (shared context, decisions, state) rather than a single person.

func FormatPeople added in v0.5.1 

func FormatPeople(people []Person, tag string) string

FormatPeople formats people list with specified XML tag. Format: [Person:ID] Name (@username, aka Alias1, Alias2) [Circle]: Bio If tag is empty, outputs plain list without XML wrapper.

func FormatRecentTopics added in v0.4.8 

func FormatRecentTopics(topics []TopicExtended) string

FormatRecentTopics formats recent topics for inclusion in agent prompts. Returns content wrapped in <recent_topics> tags. Format: - date: "summary" (N msg, ~Xk chars)

func FormatUserProfile added in v0.4.8 

func FormatUserProfile(facts []Fact) string

FormatUserProfile formats user facts for inclusion in agent prompts. Returns content wrapped in <user_profile> tags. Format: - [Fact:X] Category/Type (Updated: date) Content Use this for agents that need Fact IDs (e.g., Archivist for update/delete).

func FormatUserProfileCompact added in v0.6.1 

func FormatUserProfileCompact(facts []Fact) string

FormatUserProfileCompact formats user facts without Fact IDs. Returns content wrapped in <user_profile> tags. Format: - Category/Type (Updated: date) Content Use this for agents that don't need Fact IDs (e.g., Reranker, Laplace). This prevents ID format confusion with [Person:N], [Topic:N], [Artifact:N].

func NormalizeCircle added in v0.10.2 

func NormalizeCircle(c string) string

NormalizeCircle returns c when it is a recognized circle value, otherwise CircleOther. It guards the write paths against empty or arbitrary values coming from LLM output so the stored taxonomy stays within the known set.

func NormalizeFactKind added in v0.11.0 

func NormalizeFactKind(k string) string

NormalizeFactKind returns k when it is a recognized kind, otherwise FactKindSelfReport. It guards the write paths against empty or arbitrary values coming from LLM output so the stored taxonomy stays within the known set.

func RecordCleanupDeleted added in v0.3.5 

func RecordCleanupDeleted(table string, count int64)

RecordCleanupDeleted records the number of deleted rows during cleanup.

func RecordCleanupDuration added in v0.3.5 

func RecordCleanupDuration(table string, seconds float64)

RecordCleanupDuration records the duration of a cleanup operation.

func SetStorageSize added in v0.3.5 

func SetStorageSize(bytes int64)

SetStorageSize updates the storage size metric.

func SetTableSize added in v0.3.5 

func SetTableSize(table string, bytes int64)

SetTableSize updates the table size metric.

Types

type AgentLog added in v0.4.8 

type AgentLog struct {
	ID                int64
	UserID            ScopeID
	AgentType         string // laplace, reranker, splitter, merger, enricher, archivist, scout
	InputPrompt       string
	InputContext      string // JSON - full LLM API request
	OutputResponse    string
	OutputParsed      string // JSON - structured output
	OutputContext     string // JSON - full LLM API response
	Model             string
	PromptTokens      int
	CompletionTokens  int
	TotalCost         *float64
	DurationMs        int
	Metadata          string // JSON - agent-specific data
	Success           bool
	ErrorMessage      string
	ConversationTurns string // JSON - all request/response turns for multi-turn agents
	CreatedAt         time.Time
}

AgentLog stores debug traces from LLM agent calls (unified logging for all agents)

type AgentLogFilter added in v0.4.8 

type AgentLogFilter struct {
	UserID    ScopeID
	AgentType string
	Success   *bool
	Search    string
}

AgentLogFilter for filtering agent logs

type AgentLogRepository added in v0.4.8 

type AgentLogRepository interface {
	AddAgentLog(log AgentLog) error
	GetAgentLogs(agentType string, userID ScopeID, limit int) ([]AgentLog, error)
	GetAgentLogsExtended(filter AgentLogFilter, limit, offset int) (AgentLogResult, error)
	GetAgentLogFull(ctx context.Context, id int64, userID ScopeID) (*AgentLog, error)
}

AgentLogRepository handles unified agent debug log operations.

type AgentLogResult added in v0.4.8 

type AgentLogResult struct {
	Data       []AgentLog
	TotalCount int
}

AgentLogResult wraps agent logs with total count for pagination

type Artifact added in v0.6.0 

type Artifact struct {
	ID        int64   `json:"id"`
	UserID    ScopeID `json:"user_id"`
	MessageID int64   `json:"message_id"`

	// File metadata
	FileType     string `json:"file_type"` // 'image', 'voice', 'pdf', 'video_note', 'document'
	FilePath     string `json:"file_path"` // Relative path from storage dir
	FileSize     int64  `json:"file_size"` // Bytes
	MimeType     string `json:"mime_type"`
	OriginalName string `json:"original_name"` // From Telegram

	// Deduplication
	ContentHash string `json:"content_hash"` // SHA256 of file content

	// Processing status
	State        string  `json:"state"` // 'pending', 'processing', 'ready', 'failed'
	ErrorMessage *string `json:"error_message,omitempty"`

	// Retry tracking (v0.6.0 - CRIT-3)
	RetryCount   int        `json:"retry_count"`
	LastFailedAt *time.Time `json:"last_failed_at,omitempty"`

	// AI-generated metadata (summary-based search, populated in Phase 2)
	Summary   *string   `json:"summary,omitempty"`   // 2-4 sentence description
	Keywords  *string   `json:"keywords,omitempty"`  // JSON array: ["tag1", "tag2"]
	Entities  *string   `json:"entities,omitempty"`  // JSON array: ["person", "company"]
	RAGHints  *string   `json:"rag_hints,omitempty"` // JSON array: ["what questions?"]
	Embedding []float32 `json:"embedding,omitempty"` // Summary embedding for vector search

	// Timestamps
	CreatedAt   time.Time  `json:"created_at"`
	ProcessedAt *time.Time `json:"processed_at"`

	// Usage tracking (v0.6.0)
	ContextLoadCount int        `json:"context_load_count"` // How many times loaded into LLM context
	LastLoadedAt     *time.Time `json:"last_loaded_at"`     // Last time loaded

	// User context (v0.6.0) - text of message(s) when file was sent
	UserContext *string `json:"user_context,omitempty"`
}

Artifact represents a file stored in the artifacts system. v0.6.0: Simplified with summary-based search (no chunks, no full_text).

type ArtifactFilter added in v0.6.0 

type ArtifactFilter struct {
	UserID   ScopeID
	State    string // "pending", "processing", "ready", "failed", or "" for all
	FileType string // "image", "voice", "pdf", "video_note", "document", or "" for all
}

ArtifactFilter defines filtering options for artifact queries.

type ArtifactReferenceRepository added in v0.11.0 

type ArtifactReferenceRepository interface {
	// PersistOutboundDeliveryReplyWithArtifacts performs the same exact history,
	// transport-message, and delivery linkage as PersistOutboundDeliveryReply,
	// while assigning canonical ownership only for newly-created artifacts and
	// recording every delivered generated/stored artifact as an ordered M:N ref.
	PersistOutboundDeliveryReplyWithArtifacts(userID ScopeID, deliveryID int64, message Message, artifacts PersistOutboundArtifacts) (int64, error)
	// GetHistoryArtifactReferences returns all refs for one owned history row in
	// stable ordinal order. A foreign/missing history id yields an empty result.
	GetHistoryArtifactReferences(userID ScopeID, historyID int64) ([]HistoryArtifactReference, error)
}

ArtifactReferenceRepository is the V2 extension for provenance-preserving artifact delivery. It remains separate from DeliveryRepository so existing focused implementations and call sites stay source compatible while new delivery paths can atomically persist repeated stored-artifact references.

type ArtifactReferenceSource added in v0.11.0 

type ArtifactReferenceSource string

ArtifactReferenceSource records why an existing artifact was associated with an assistant reply. It is intentionally bounded: callers may either persist an artifact generated by the current turn or one selected from the user's trusted stored-artifact inventory. 

const (
	ArtifactReferenceSourceGenerated ArtifactReferenceSource = "generated"
	ArtifactReferenceSourceStored    ArtifactReferenceSource = "stored"
)

type ArtifactRepository added in v0.6.0 

type ArtifactRepository interface {
	AddArtifact(artifact Artifact) (int64, error)
	GetArtifact(userID ScopeID, artifactID int64) (*Artifact, error)
	GetByHash(userID ScopeID, contentHash string) (*Artifact, error)
	// GetPendingArtifacts returns artifacts ready for processing:
	// - state='pending' (new artifacts)
	// - state='failed' with retry_count < maxRetries and sufficient backoff elapsed (v0.6.0)
	GetPendingArtifacts(userID ScopeID, maxRetries int) ([]Artifact, error)
	GetArtifacts(filter ArtifactFilter, limit, offset int) ([]Artifact, int64, error)
	UpdateArtifact(artifact Artifact) error
	RecoverArtifactStates(threshold time.Duration) error
	GetArtifactsByIDs(userID ScopeID, artifactIDs []int64) ([]Artifact, error)
	// GetSessionArtifacts returns artifacts on messages still in active session (topic_id IS NULL).
	// Used to inject freshly-created artifacts as priority candidates for the reranker.
	GetSessionArtifacts(ctx context.Context, userID ScopeID, limit int, maxAge time.Duration) ([]Artifact, error)
	// IncrementContextLoadCount tracks usage when artifacts are loaded into LLM context (v0.6.0)
	IncrementContextLoadCount(userID ScopeID, artifactIDs []int64) error
	// UpdateMessageID links artifact to history message (called after message is saved)
	// Requires userID for proper data isolation.
	UpdateMessageID(userID ScopeID, artifactID, messageID int64) error

	// v0.7.0: embedding migration
	GetArtifactsNeedingReembed(expectedVersion string, limit int) ([]ReembedCandidate, error)
	UpdateArtifactEmbeddingVersion(id int64, emb []float32, version string) error
}

ArtifactRepository handles artifact file metadata operations.

type Channel added in v0.10.0 

type Channel struct {
	ScopeID     ScopeID
	Transport   string
	NativeID    string
	DisplayName string
	CreatedAt   time.Time
	LastSeen    time.Time
}

Channel is a channel scope: a conversation with many participants, as opposed to a person scope. Keyed deterministically by (transport, native_id); participants live as people within the scope.

type ChannelRepository added in v0.10.0 

type ChannelRepository interface {
	GetOrCreateChannel(transport, nativeID, displayName string) (ScopeID, error)
	GetChannel(scopeID ScopeID) (*Channel, error)
}

ChannelRepository manages channel scopes. A channel scope is keyed deterministically by (transport, native_id); participants are tracked as people within the scope.

type CheckpointResult added in v0.4.6 

type CheckpointResult struct {
	Busy         int // 0 = success, 1 = blocked by reader
	Log          int // Total frames in WAL file
	Checkpointed int // Frames actually checkpointed
}

CheckpointResult contains the result of a WAL checkpoint operation.

type Config added in v0.10.0 

type Config struct {
	Driver   string // "sqlite" (default) | "postgres"
	Path     string // SQLite file path
	Postgres PostgresConfig
}

Config selects and parameterizes the storage backend.

type ContaminatedTopic added in v0.4.6 

type ContaminatedTopic struct {
	TopicID       int64     `json:"topic_id"`
	TopicOwner    ScopeID   `json:"topic_owner"`
	TopicSummary  string    `json:"topic_summary"`
	ForeignUsers  []ScopeID `json:"foreign_users"`
	ForeignMsgCnt int       `json:"foreign_msg_count"`
	TotalMsgCnt   int       `json:"total_msg_count"`
}

ContaminatedTopic represents a topic containing messages from other users.

type DashboardStats 

type DashboardStats struct {
	TotalTopics         int
	AvgTopicSize        float64
	ProcessedTopicsPct  float64
	ConsolidatedTopics  int
	TotalFacts          int
	FactsByCategory     map[string]int
	FactsByType         map[string]int
	TotalMessages       int
	UnprocessedMessages int
	TotalRAGQueries     int
	AvgRAGCost          float64
	MessagesPerDay      map[string]int
	FactsGrowth         map[string]int
}

type DeliveryErrorClass added in v0.11.0 

type DeliveryErrorClass string

DeliveryErrorClass is intentionally bounded. The ledger must never receive a raw error string because it may contain a URL, response fragment, or content. 

const (
	DeliveryErrorNone            DeliveryErrorClass = ""
	DeliveryErrorFormat          DeliveryErrorClass = "format"
	DeliveryErrorRateLimit       DeliveryErrorClass = "rate_limit"
	DeliveryErrorNetwork         DeliveryErrorClass = "network"
	DeliveryErrorServer          DeliveryErrorClass = "server"
	DeliveryErrorInvalidResponse DeliveryErrorClass = "invalid_response"
	DeliveryErrorInternal        DeliveryErrorClass = "internal"
	DeliveryErrorInterrupted     DeliveryErrorClass = "interrupted"
)

type DeliveryOperationKind added in v0.11.0 

type DeliveryOperationKind string

DeliveryOperationKind mirrors the closed persistent-operation union in the delivery planner. Keeping this bounded prevents arbitrary caller strings from turning the content-free ledger into an accidental logging channel. 

const (
	DeliveryOperationRichText   DeliveryOperationKind = "rich_text"
	DeliveryOperationLegacyText DeliveryOperationKind = "legacy_text"
	DeliveryOperationRichMedia  DeliveryOperationKind = "rich_media"
	DeliveryOperationMedia      DeliveryOperationKind = "media"
)

type DeliveryOperationStatus added in v0.11.0 

type DeliveryOperationStatus string
const (
	DeliveryOperationStatusPlanned        DeliveryOperationStatus = "planned"
	DeliveryOperationStatusSending        DeliveryOperationStatus = "sending"
	DeliveryOperationStatusConfirmed      DeliveryOperationStatus = "confirmed"
	DeliveryOperationStatusRejected       DeliveryOperationStatus = "rejected"
	DeliveryOperationStatusFormatRejected DeliveryOperationStatus = "format_rejected"
	DeliveryOperationStatusSkipped        DeliveryOperationStatus = "skipped"
	DeliveryOperationStatusUnknown        DeliveryOperationStatus = "unknown"
)

type DeliveryRepository added in v0.11.0 

type DeliveryRepository interface {
	CreateOutboundDelivery(delivery OutboundDelivery, operations []OutboundDeliveryOperation) (int64, error)
	MarkOutboundDeliveryOperationSending(deliveryID int64, ordinal int) error
	CompleteOutboundDeliveryOperation(deliveryID int64, ordinal int, status DeliveryOperationStatus, errorClass DeliveryErrorClass, transportMessageIDs []string) error
	ActivateOutboundDeliveryFallback(deliveryID int64, rejectedOrdinal int, operations []OutboundDeliveryOperation) ([]int, error)
	MarkInterruptedOutboundDeliveriesUnknown() (int64, error)
	GetOutboundDelivery(deliveryID int64) (*OutboundDelivery, []OutboundDeliveryOperation, error)
	// GetOutboundDeliveryByTransportMessage resolves a confirmed operation's
	// content-free parent delivery by its exact transport identity. It supports
	// reactions during the bounded window before (or without) history linkage.
	GetOutboundDeliveryByTransportMessage(userID ScopeID, transport, conversationID, transportMsgID string) (*OutboundDelivery, error)
	LinkOutboundDeliveryHistory(userID ScopeID, deliveryID, historyID int64) error
	// PersistOutboundDeliveryReply atomically inserts an assistant history row,
	// links every confirmed transport id, associates generated artifacts, and
	// records the history id on the delivery.
	PersistOutboundDeliveryReply(userID ScopeID, deliveryID int64, message Message, artifactIDs []int64) (int64, error)
}

DeliveryRepository persists the content-free state machine for persistent outbound operations. It exists independently of MessageRepository so callers that only need conversation history do not acquire delivery responsibilities.

type DeliveryStatus added in v0.11.0 

type DeliveryStatus string
const (
	DeliveryStatusPlanned         DeliveryStatus = "planned"
	DeliveryStatusSending         DeliveryStatus = "sending"
	DeliveryStatusConfirmed       DeliveryStatus = "confirmed"
	DeliveryStatusRejected        DeliveryStatus = "rejected"
	DeliveryStatusPartialRejected DeliveryStatus = "partial_rejected"
	DeliveryStatusUnknown         DeliveryStatus = "unknown"
	DeliveryStatusPartialUnknown  DeliveryStatus = "partial_unknown"
)

type Dialect added in v0.10.0 

type Dialect interface {
	// Name reports the backend ("sqlite" | "postgres").
	Name() string

	// Rebind converts `?` placeholders into the dialect's parameter syntax.
	// SQLite keeps `?`; Postgres numbers them left-to-right ($1, $2, ...).
	Rebind(query string) string

	// BindTime returns the value to bind for a timestamp column. SQLite stores
	// the canonical "2006-01-02 15:04:05.999" UTC string (preserving the exact
	// original SQLite representation); Postgres binds time.Time for native timestamptz.
	BindTime(t time.Time) any

	// SinceDaysPredicate renders a "<col> >= <now − days>" predicate.
	SinceDaysPredicate(col string, days int) string

	// DateExpr renders day-truncation of a timestamp column (for GROUP BY day).
	DateExpr(col string) string

	// AvgAgeDaysExpr renders AVG age-in-days of <col> relative to now.
	AvgAgeDaysExpr(col string) string

	// MinutesAgoExpr renders the timestamp "<now − minutes>" as a SQL expression
	// (no bound parameter) for inline retry-backoff predicates.
	MinutesAgoExpr(minutes int) string

	// SecondsAgoExpr renders "<col> < <now − ? seconds>", where the seconds
	// count is supplied by a single bound `?` parameter at the call site.
	SecondsAgoExpr(col string) string

	// BoolLit renders a boolean literal for WHERE comparisons (SQLite: 0/1,
	// Postgres: FALSE/TRUE).
	BoolLit(b bool) string
}

Dialect abstracts the handful of SQL constructs that genuinely differ between the SQLite and PostgreSQL backends. Everything else — table shapes, ON CONFLICT upserts, CURRENT_TIMESTAMP defaults — is written once and shared. The SQLite dialect is a behavioral no-op: its generated SQL stays byte-identical to the pre-dialect code, so the existing SQLite deployment is unaffected.

Ordering rule: ExpandIn (sqlin.go) expands `(?)` → `(?,?,?)` FIRST; only then does Rebind number the placeholders. Never Rebind before ExpandIn.

type ExactMessageRepository added in v0.11.0 

type ExactMessageRepository interface {
	// AddMessageToHistoryReturningID inserts one row and returns its exact id.
	// Multi-part delivery paths must use this instead of guessing "the latest"
	// assistant row when linking transport ids and generated artifacts.
	AddMessageToHistoryReturningID(userID ScopeID, message Message) (int64, error)
	// LinkReplyTransportMessages associates every persistent transport message
	// produced by one logical reply with the exact history row.
	LinkReplyTransportMessages(userID ScopeID, historyID int64, messages []TransportMessage) error
	// GetReplyByTransportMessage resolves the collision-safe composite identity.
	// It dual-reads the normalized mapping and pre-v19 history attribution.
	GetReplyByTransportMessage(userID ScopeID, transport, conversationID, transportMsgID string) (*Message, error)
}

ExactMessageRepository is the V2 extension for collision-safe, exact-row reply persistence. It remains separate from MessageRepository so older consumers and focused test doubles are source compatible.

type Fact 

type Fact struct {
	ID          int64
	UserID      ScopeID
	Relation    string
	Content     string
	Category    string
	Type        string // identity, context, status
	Kind        string // provenance: self_report, user_opinion, verified, constraint (see factkind.go)
	Importance  int    // 0-100
	Embedding   []float32
	TopicID     *int64 // Nullable
	CreatedAt   time.Time
	LastUpdated time.Time
}

func FilterProfileFacts added in v0.4.8 

func FilterProfileFacts(facts []Fact) []Fact

FilterProfileFacts filters facts to identity and high-importance facts only. This is the standard filter used across all agents. Constraint facts are always included regardless of importance: a stored behavioral restriction the model never sees is silently either violated or (worse) half-remembered — it must reach the prompt to be weighed at all.

type FactHistory 

type FactHistory struct {
	ID           int64
	FactID       int64
	UserID       ScopeID
	Action       string // add, update, delete
	OldContent   string
	NewContent   string
	Reason       string
	Category     string
	Relation     string
	Importance   int
	TopicID      *int64
	CreatedAt    time.Time
	RequestInput string
}

type FactHistoryFilter 

type FactHistoryFilter struct {
	UserID   ScopeID
	Action   string
	Category string
	Search   string
}

type FactHistoryRepository 

type FactHistoryRepository interface {
	AddFactHistory(history FactHistory) error
	UpdateFactHistoryTopic(oldTopicID, newTopicID int64) error
	GetFactHistory(userID ScopeID, limit int) ([]FactHistory, error)
	GetFactHistoryExtended(filter FactHistoryFilter, limit, offset int, sortBy, sortDir string) (FactHistoryResult, error)
}

FactHistoryRepository handles fact history operations.

type FactHistoryResult 

type FactHistoryResult struct {
	Data       []FactHistory
	TotalCount int
}

type FactRepository 

type FactRepository interface {
	AddFact(fact Fact) (int64, error)
	GetFacts(userID ScopeID) ([]Fact, error)
	GetFactsByIDs(userID ScopeID, ids []int64) ([]Fact, error)
	GetFactsByTopicID(userID ScopeID, topicID int64) ([]Fact, error)
	GetAllFacts() ([]Fact, error)
	GetFactsAfterID(minID int64) ([]Fact, error)
	GetFactStats() (FactStats, error)
	GetFactStatsByUser(userID ScopeID) (FactStats, error)
	UpdateFact(fact Fact) error
	UpdateFactsTopic(userID ScopeID, oldTopicID, newTopicID int64) error
	DeleteFact(userID ScopeID, id int64) error

	// v0.7.0: embedding migration
	GetFactsNeedingReembed(expectedVersion string, limit int) ([]ReembedCandidate, error)
	UpdateFactEmbeddingVersion(id int64, emb []float32, version string) error
}

FactRepository handles fact operations.

Facts are structured pieces of information extracted from conversations by the Archivist agent. They form the user's long-term profile memory.

Fact Types:

  • identity: Core user information (name, location)
  • importance: User-defined importance score (0-100)
  • Facts with importance ≥ 90 are always included in profile

Each fact has an embedding vector for semantic search and deduplication.

type FactStats 

type FactStats struct {
	CountByType map[string]int
	AvgAgeDays  float64
}

type Flag added in v0.10.2 

type Flag struct {
	ID           int64
	UserID       ScopeID
	HistoryID    *int64  // history.id of the flagged assistant reply (nullable)
	MessageID    string  // transport-native message id the user reacted to
	TraceID      *string // trace that produced the reply (nullable)
	Emoji        string  // the reaction emoji
	ReplyPreview string  // truncated copy of the reply content
	CreatedAt    time.Time
}

Flag is one reaction a user added to a bot reply, marking it as a bad response (migration 016). It carries the trace_id of the reply so the operator can jump straight to the trace that produced it.

type FlagRepository added in v0.10.2 

type FlagRepository interface {
	AddFlag(flag Flag) error
	GetFlags(userID ScopeID, limit int) ([]Flag, error)
}

FlagRepository handles user-flagged bad replies (migration 016). A flag is recorded when a user reacts to a bot reply; it carries the reply's trace_id so the operator can investigate straight from the trace.

type HistoryArtifactReference added in v0.11.0 

type HistoryArtifactReference struct {
	HistoryID  int64
	UserID     ScopeID
	ArtifactID int64
	Ordinal    int
	Mode       artifactdelivery.Mode
	Source     ArtifactReferenceSource
	CreatedAt  time.Time
}

HistoryArtifactReference is the persisted M:N association returned in delivery order.

type Identity added in v0.10.0 

type Identity struct {
	Transport string
	NativeID  string
	ScopeID   ScopeID
	CreatedAt time.Time
	LastSeen  time.Time
}

Identity maps a transport-native handle to its scope. Many identities may point at one principal scope for unified cross-transport memory.

type IdentityRepository added in v0.10.0 

type IdentityRepository interface {
	GetIdentity(transport, nativeID string) (*Identity, error)
	PutIdentity(transport, nativeID string, scopeID ScopeID) error
}

IdentityRepository maps a transport-native handle to its scope id. The resolver-driven DM flow looks up an identity to reuse an existing scope, then writes one after resolving a principal. Telegram passthrough writes no identity row (its id is a deterministic uuidv5).

type MaintenanceRepository added in v0.3.5 

type MaintenanceRepository interface {
	GetDBSize() (int64, error)
	GetTableSizes() ([]TableSize, error)
	CleanupFactHistory(keepPerUser int) (int64, error)
	CleanupAgentLogs(keepPerUserPerAgent int, minAge time.Duration) (int64, error)
	CountAgentLogs() (int64, error)
	CountFactHistory() (int64, error)

	// Database health diagnostics
	CountOrphanedTopics(userID ScopeID) (int, error)
	GetOrphanedTopicIDs(userID ScopeID) ([]int64, error)
	CountOverlappingTopics(userID ScopeID) (int, error)
	GetOverlappingTopics(userID ScopeID) ([]OverlappingPair, error)
	CountFactsOnOrphanedTopics(userID ScopeID) (int, error)
	RecalculateTopicRanges(userID ScopeID) (int, error)
	RecalculateTopicSizes(userID ScopeID) (int, error)

	// Cross-user contamination detection and repair
	GetContaminatedTopics(userID ScopeID) ([]ContaminatedTopic, error)
	CountContaminatedTopics(userID ScopeID) (int, error)
	FixContaminatedTopics(userID ScopeID) (int64, error)

	// WAL checkpoint for ensuring data persistence
	Checkpoint() error
}

MaintenanceRepository handles database maintenance operations.

type MemoryBankRepository 

type MemoryBankRepository interface {
	GetMemoryBank(userID ScopeID) (string, error)
	UpdateMemoryBank(userID ScopeID, content string) error
}

MemoryBankRepository handles the legacy memory bank.

type MergeCandidate 

type MergeCandidate struct {
	Topic1 Topic
	Topic2 Topic
}

type Message 

type Message struct {
	ID        int64
	UserID    ScopeID
	Role      string
	Content   string
	TopicID   *int64 // Nullable
	CreatedAt time.Time

	// Multi-transport attribution (v0.10, migration 012). Nullable; unused on
	// the single-user Telegram/Mattermost DM paths. Populated only by
	// multi-participant transports for channel attribution / edits / reactions.
	Author         *string // author display/handle within the scope
	MessageID      *string // transport-native message/post id
	ConversationID *string // transport-native chat/channel id

	// ThreadRoot is the transport thread this message belongs to (migration
	// 013), recorded on channel messages and the bot's channel replies. Reply
	// gating uses the post's quote (ReplyToBot), not thread membership; this
	// column is kept as thread-membership metadata for thread-scoped context.
	// NULL in DMs.
	ThreadRoot *string

	// TraceID is the trace that produced this row (migration 016). Set on
	// assistant replies so an inbound reaction on the reply resolves to its
	// trace. NULL on user rows and on replies stored before migration 016.
	TraceID *string

	// DoNotStore marks a message written while the scope's privacy mode was
	// on (migration 018). The row stays in raw history for short-term session
	// context, but its content is excluded from long-term memory: the topic
	// pipeline redacts it before the splitter/embeddings, and
	// GetMessagesByTopicID filters it out (archivist, RAG re-injection,
	// merger, topic views).
	DoNotStore bool
}

type MessageRepository 

type MessageRepository interface {
	AddMessageToHistory(userID ScopeID, message Message) error
	ImportMessage(userID ScopeID, message Message) error
	GetRecentHistory(userID ScopeID, limit int) ([]Message, error)
	GetMessagesByIDs(userID ScopeID, ids []int64) ([]Message, error)
	ClearHistory(userID ScopeID) error
	GetMessagesInRange(ctx context.Context, userID ScopeID, startID, endID int64) ([]Message, error)
	GetMessagesByTopicID(ctx context.Context, topicID int64) ([]Message, error)
	UpdateMessageTopic(userID ScopeID, messageID, topicID int64) error
	UpdateMessagesTopicInRange(ctx context.Context, userID ScopeID, startMsgID, endMsgID, topicID int64) error
	GetUnprocessedMessages(userID ScopeID) ([]Message, error)
	GetRecentSessionMessages(ctx context.Context, userID ScopeID, limit int, excludeIDs []int64) ([]Message, error)
	// SetReplyTransportID back-fills the transport-native message id on the user's
	// most recent unlinked assistant reply, after that reply is sent.
	SetReplyTransportID(userID ScopeID, transportMsgID string) error
	// GetReplyByTransportID resolves an assistant reply by its transport-native
	// message id (nil, nil on miss). Used by the inbound-reaction handler.
	GetReplyByTransportID(userID ScopeID, transportMsgID string) (*Message, error)
}

MessageRepository handles message history operations.

Messages represent the conversation log between user and assistant. Message IDs are globally auto-incremented across all users.

Critical: Any range query MUST include user_id filter to prevent data leakage.

type OutboundArtifactReference added in v0.11.0 

type OutboundArtifactReference struct {
	ArtifactID int64
	Ordinal    int
	Mode       artifactdelivery.Mode
	Source     ArtifactReferenceSource
}

OutboundArtifactReference is one ordered, transport-neutral artifact presentation attached to a logical assistant reply. Repeating ArtifactID is valid (for example preview followed by the byte-exact original); Ordinal is the stable ordering key.

type OutboundDelivery added in v0.11.0 

type OutboundDelivery struct {
	ID             int64
	UserID         ScopeID
	Transport      string
	ConversationID string
	TraceID        *string
	HistoryID      *int64
	Status         DeliveryStatus
	OperationCount int
	ConfirmedCount int
	CreatedAt      time.Time
	UpdatedAt      time.Time
}

type OutboundDeliveryOperation added in v0.11.0 

type OutboundDeliveryOperation struct {
	ID                  int64
	DeliveryID          int64
	Ordinal             int
	Kind                DeliveryOperationKind
	Status              DeliveryOperationStatus
	ErrorClass          DeliveryErrorClass
	StartedAt           *time.Time
	FinishedAt          *time.Time
	CreatedAt           time.Time
	TransportMessageIDs []string
}

type OverlappingPair added in v0.4.6 

type OverlappingPair struct {
	Topic1ID      int64
	Topic1Summary string
	Topic2ID      int64
	Topic2Summary string
}

OverlappingPair contains information about two overlapping topics.

type PeopleRepository added in v0.5.1 

type PeopleRepository interface {
	// CRUD operations
	AddPerson(person Person) (int64, error)
	UpdatePerson(person Person) error
	DeletePerson(userID ScopeID, personID int64) error

	// Retrieval
	GetPerson(userID ScopeID, personID int64) (*Person, error)
	GetPeople(userID ScopeID) ([]Person, error)
	GetPeopleByIDs(userID ScopeID, ids []int64) ([]Person, error)
	GetAllPeople() ([]Person, error)
	GetPeopleAfterID(minID int64) ([]Person, error)

	// Direct matching (fast path for @username and name lookup)
	FindPersonByTelegramID(userID ScopeID, telegramID int64) (*Person, error)
	// FindPersonByExternalID matches on the transport-neutral external id
	// (transport, native_id) introduced in v0.10. For Telegram this is
	// equivalent to FindPersonByTelegramID via the backfilled ('telegram', id).
	FindPersonByExternalID(userID ScopeID, transport, nativeID string) (*Person, error)
	FindPersonByUsername(userID ScopeID, username string) (*Person, error)
	FindPersonByAlias(userID ScopeID, alias string) ([]Person, error)
	FindPersonByName(userID ScopeID, name string) (*Person, error)

	// Merge operations
	MergePeople(userID ScopeID, targetID, sourceID int64, newBio string, newAliases []string, newUsername *string, newTelegramID *int64) error

	// Extended queries with filtering and pagination
	GetPeopleExtended(filter PersonFilter, limit, offset int, sortBy, sortDir string) (PersonResult, error)

	// Maintenance
	CountPeopleWithoutEmbedding(userID ScopeID) (int, error)
	GetPeopleWithoutEmbedding(userID ScopeID) ([]Person, error)

	// v0.7.0: embedding migration
	GetPeopleNeedingReembed(expectedVersion string, limit int) ([]ReembedCandidate, error)
	UpdatePersonEmbeddingVersion(id int64, emb []float32, version string) error
}

PeopleRepository handles people from the user's social graph.

type PersistOutboundArtifacts added in v0.11.0 

type PersistOutboundArtifacts struct {
	OwnedArtifactIDs []int64
	References       []OutboundArtifactReference
}

PersistOutboundArtifacts separates provenance ownership from repeated use. OwnedArtifactIDs are newly-created rows whose temporary message_id=0 should become the new history row. References associate delivered artifacts with the reply without changing their canonical artifacts.message_id owner.

type Person added in v0.5.1 

type Person struct {
	ID           int64     `json:"id"`
	UserID       ScopeID   `json:"user_id"`
	DisplayName  string    `json:"display_name"`
	Aliases      []string  `json:"aliases"`     // JSON array: ["Johnny", "@johndoe"]
	TelegramID   *int64    `json:"telegram_id"` // For direct @mention match
	Username     *string   `json:"username"`    // @username without @
	Circle       string    `json:"circle"`      // Family, Friends, Work_Inner, Work_Outer, Other
	Bio          string    `json:"bio"`         // Aggregated profile (2-3 sentences)
	Embedding    []float32 `json:"embedding"`   // Bio vector (JSON float32 array)
	FirstSeen    time.Time `json:"first_seen"`
	LastSeen     time.Time `json:"last_seen"`
	MentionCount int       `json:"mention_count"`

	// External identity (migration 011): transport-neutral (transport, native_id)
	// for non-Telegram participants such as channel members. Telegram people are
	// backfilled to ('telegram', telegram_id). Nil when unset.
	ExternalTransport *string `json:"external_transport,omitempty"`
	ExternalID        *string `json:"external_id,omitempty"`
}

Person represents a person from the user's social graph.

func FilterInnerCircle added in v0.5.1 

func FilterInnerCircle(people []Person) []Person

FilterInnerCircle returns only Work_Inner and Family people.

type PersonFilter added in v0.5.1 

type PersonFilter struct {
	UserID ScopeID
	Circle string
	Search string
}

PersonFilter for filtering people queries.

type PersonResult added in v0.5.1 

type PersonResult struct {
	Data       []Person
	TotalCount int
}

PersonResult wraps people with total count for pagination.

type PostgresConfig added in v0.10.0 

type PostgresConfig struct {
	Host     string
	Port     int
	Database string
	User     string
	Password string
	SSLMode  string
}

PostgresConfig holds connection parameters for the postgres backend.

type Principal added in v0.10.0 

type Principal struct {
	ScopeID     ScopeID
	ObjectGUID  string // AD anchor; "" until a Keycloak/LDAP lookup lands
	ADLogin     string // lowercase preferred_username; the external join key
	Email       string
	DisplayName string
	CreatedAt   time.Time
	UpdatedAt   time.Time
}

Principal is an AD-backed person scope. One principal is one human; many transport identities may map to a single principal scope so memory follows the person across transports.

type PrincipalInput added in v0.10.0 

type PrincipalInput struct {
	ObjectGUID  string
	ADLogin     string
	Email       string
	DisplayName string
}

PrincipalInput carries the resolved attributes used to dedup-or-create a principal. Dedup prefers ObjectGUID (stable across logon renames) and falls back to ADLogin. Empty fields are stored as NULL so the partial unique indexes on object_guid / ad_login do not treat unknown values as collisions.

type PrincipalRepository added in v0.10.0 

type PrincipalRepository interface {
	GetOrCreatePrincipal(in PrincipalInput) (ScopeID, bool, error)
	GetPrincipal(scopeID ScopeID) (*Principal, error)
}

PrincipalRepository manages AD-backed person scopes. A principal is one human; many transport identities may map to a single principal scope for unified cross-transport memory. Dedup prefers object_guid (the stable AD anchor, nullable until a later lookup) and falls back to ad_login (lowercase preferred_username), so a later object_guid backfill is additive and never re-partitions memory.

type ReembedCandidate added in v0.7.0 

type ReembedCandidate struct {
	ID      int64
	UserID  ScopeID
	Content string
}

ReembedCandidate identifies a row that needs its embedding re-generated because the current `embedding_version` does not match the configured embedding model + dimension.

Content is the text that should be passed to the embedding model for this entity (summary for topics/artifacts, body for facts, composed searchable string for people).

type RerankerCandidate added in v0.4.1 

type RerankerCandidate struct {
	TopicID      int64   `json:"topic_id"`
	Summary      string  `json:"summary"`
	Score        float32 `json:"score"`
	Date         string  `json:"date"`
	MessageCount int     `json:"message_count"`
	SizeChars    int     `json:"size_chars"`
}

RerankerCandidate is a single candidate for JSON serialization

type RerankerToolCall added in v0.4.1 

type RerankerToolCall struct {
	Iteration int                     `json:"iteration"`
	TopicIDs  []int64                 `json:"topic_ids"`
	Topics    []RerankerToolCallTopic `json:"topics"`
}

RerankerToolCall represents one iteration of tool calls

type RerankerToolCallTopic added in v0.4.1 

type RerankerToolCallTopic struct {
	ID      int64  `json:"id"`
	Summary string `json:"summary"`
}

RerankerToolCallTopic contains topic info for tool call display

type ScopeID added in v0.10.0 

type ScopeID string

ScopeID is the memory partition key (tenant): an opaque UUID identifying a scope — a person (principal), a channel, or a passthrough identity. It replaces the former int64 user_id as the partition key throughout storage.

Representation: a UUID rendered as the canonical 36-char lowercase string. On Postgres the physical column is `uuid`; on SQLite it is TEXT. The physical column is still named `user_id` (renaming it across ~180 shared SQL statements is deferred) — only the Go type and the stored values change.

IMPORTANT: only the *partition* key is a ScopeID. Entity ids — message, topic, fact, person, artifact ids, and `people.telegram_id` — remain int64.

func MintScopeID added in v0.10.0 

func MintScopeID() ScopeID

MintScopeID generates a fresh random scope id for a principal — a scope that spans many transport identities and is therefore not derivable from any single (transport, native_id). Principal dedup happens via the principals table (object_guid / ad_login lookup), so the id itself need not be deterministic.

func PassthroughScopeID added in v0.10.0 

func PassthroughScopeID(transport, nativeID string) ScopeID

PassthroughScopeID derives the deterministic scope id for a transport-native identity that has no principal resolution — the Telegram path and any transport without a configured PrincipalResolver. uuidv5 makes it stable and lookup-free: the same (transport, nativeID) always maps to the same scope, so no scopes/identities row is needed to recover it.

The one-off Telegram int64→UUID migration MUST use this same function with transport "telegram" and the decimal id, or existing memory orphans.

Channel scopes also use this (a channel is keyed deterministically by its (transport, native_id)); only principal scopes get a freshly minted id.

func (ScopeID) IsZero added in v0.10.0 

func (s ScopeID) IsZero() bool

IsZero reports whether the scope id is unset.

func (*ScopeID) Scan added in v0.10.0 

func (s *ScopeID) Scan(src any) error

Scan implements sql.Scanner. SQLite's INTEGER affinity coerces numeric scope strings (e.g. "123") to integers on storage, so a scope column may scan back as int64; UUID scopes scan as string/[]byte. Handle all three so round-trips are lossless on both SQLite and Postgres (uuid → string).

func (ScopeID) String added in v0.10.0 

func (s ScopeID) String() string

String returns the underlying UUID string (handy for metric labels / JSON).

func (ScopeID) Value added in v0.10.0 

func (s ScopeID) Value() (driver.Value, error)

Value implements driver.Valuer so ScopeID is bound as a string parameter.

type ScopeRepository added in v0.10.0 

type ScopeRepository interface {
	// IsChannelScope reports whether the scope id is a multi-participant channel
	// (scope_type='channel'). Background memory loops have only the scope id and use
	// this to gate channel-aware behaviour. Absence of a row means a DM/person scope
	// (Telegram passthrough / Mattermost DM / principal), so it returns false.
	IsChannelScope(id ScopeID) (bool, error)
}

ScopeRepository exposes scope-type detection over the memory partition key. Scope creation and lookup are owned by the Identity/Principal/Channel repositories and PassthroughScopeID; this is the read side used by background loops. See internal/bot/identity.go.

type Stat 

type Stat struct {
	UserID     ScopeID
	TokensUsed int
	CostUSD    float64
}

type StatsRepository 

type StatsRepository interface {
	AddStat(stat Stat) error
	GetStats() (map[ScopeID]Stat, error)
	GetDashboardStats(userID ScopeID) (*DashboardStats, error)
}

StatsRepository handles usage statistics.

type Storage 

type Storage interface {
	MessageRepository
	ExactMessageRepository
	DeliveryRepository
	ArtifactReferenceRepository
	UserRepository
	StatsRepository
	TopicRepository
	MemoryBankRepository
	FactRepository
	FactHistoryRepository
	MaintenanceRepository
	AgentLogRepository
	PeopleRepository
	ArtifactRepository
	FlagRepository
}

type Store added in v0.10.0 

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

Store is the dialect-aware storage backend. A single codebase serves both SQLite and PostgreSQL; the active dialect is selected at construction and the SQLite dialect is a behavioral no-op so the existing SQLite deployment is unaffected.

func NewSQLiteStore 

func NewSQLiteStore(logger *slog.Logger, path string) (*Store, error)

NewSQLiteStore is a convenience constructor for the SQLite backend from a file path (used by the testbot and tests). Equivalent to NewStore(Config{Driver: "sqlite", Path: path}, logger).

func NewStore added in v0.10.0 

func NewStore(cfg Config, logger *slog.Logger) (*Store, error)

NewStore constructs the storage backend selected by cfg.Driver. An empty driver defaults to SQLite.

func (*Store) ActivateOutboundDeliveryFallback added in v0.11.0 

func (s *Store) ActivateOutboundDeliveryFallback(deliveryID int64, rejectedOrdinal int, operations []OutboundDeliveryOperation) ([]int, error)

ActivateOutboundDeliveryFallback records the one safe representation switch: a confirmed rich-format rejection to a fallback branch that was fully preflighted before delivery began. The rejected primary op and unsent primary suffix stay in the ledger as terminal audit records; fallback operations get fresh monotonically increasing ordinals.

func (*Store) AddAgentLog added in v0.10.0 

func (s *Store) AddAgentLog(log AgentLog) error

AddAgentLog inserts a new agent log entry.

func (*Store) AddArtifact added in v0.10.0 

func (s *Store) AddArtifact(artifact Artifact) (int64, error)

AddArtifact saves a new artifact to the database. Returns the ID of the inserted artifact. If an artifact with the same content_hash exists for the user, returns existing artifact ID.

func (*Store) AddFact added in v0.10.0 

func (s *Store) AddFact(fact Fact) (int64, error)

func (*Store) AddFactHistory added in v0.10.0 

func (s *Store) AddFactHistory(h FactHistory) error

func (*Store) AddFlag added in v0.10.2 

func (s *Store) AddFlag(flag Flag) error

AddFlag records a response flag.

func (*Store) AddMessageToHistory added in v0.10.0 

func (s *Store) AddMessageToHistory(userID ScopeID, message Message) error

func (*Store) AddMessageToHistoryReturningID added in v0.11.0 

func (s *Store) AddMessageToHistoryReturningID(userID ScopeID, message Message) (int64, error)

AddMessageToHistoryReturningID inserts a history row and returns that exact row's id. This is the race-free primitive for associating a delivered reply; callers must not rediscover the row via recent-history ordering.

func (*Store) AddPerson added in v0.10.0 

func (s *Store) AddPerson(person Person) (int64, error)

AddPerson creates a new person record. Returns the new person ID.

func (*Store) AddStat added in v0.10.0 

func (s *Store) AddStat(stat Stat) error

func (*Store) AddTopic added in v0.10.0 

func (s *Store) AddTopic(topic Topic) (int64, error)

func (*Store) AddTopicWithoutMessageUpdate added in v0.10.0 

func (s *Store) AddTopicWithoutMessageUpdate(topic Topic) (int64, error)

AddTopicWithoutMessageUpdate creates a topic without updating message references. Used when manually managing message-topic relationships (e.g., during topic splitting).

func (*Store) Checkpoint added in v0.10.0 

func (s *Store) Checkpoint() error

Checkpoint forces a WAL checkpoint to flush all pending writes to the main database file. This is useful before shutdown or after critical writes to ensure data persistence. Returns CheckpointResult with details about what was checkpointed.

func (*Store) CleanupAgentLogs added in v0.10.0 

func (s *Store) CleanupAgentLogs(keepPerUserPerAgent int, minAge time.Duration) (int64, error)

CleanupAgentLogs removes old agent_logs records, keeping the N most recent per user per agent type. Rows younger than minAge are kept regardless of count, so an active user's burst of calls cannot flush the debugging window within hours (agent_logs is the only place full prompts/responses survive; trace retention is much shorter). minAge <= 0 disables the age protection and trims purely by count (the purge path relies on this with keep=0). Returns the number of deleted rows.

func (*Store) CleanupFactHistory added in v0.10.0 

func (s *Store) CleanupFactHistory(keepPerUser int) (int64, error)

CleanupFactHistory removes old fact_history records, keeping only the N most recent per user. Returns the number of deleted rows.

func (*Store) ClearHistory added in v0.10.0 

func (s *Store) ClearHistory(userID ScopeID) error

func (*Store) Close added in v0.10.0 

func (s *Store) Close() error

func (*Store) CompleteOutboundDeliveryOperation added in v0.11.0 

func (s *Store) CompleteOutboundDeliveryOperation(deliveryID int64, ordinal int, status DeliveryOperationStatus, errorClass DeliveryErrorClass, transportMessageIDs []string) error

func (*Store) CountAgentLogs added in v0.10.0 

func (s *Store) CountAgentLogs() (int64, error)

CountAgentLogs returns the total number of agent_logs records.

func (*Store) CountContaminatedTopics added in v0.10.0 

func (s *Store) CountContaminatedTopics(userID ScopeID) (int, error)

CountContaminatedTopics counts topics with cross-user contamination. If userID is 0, counts all; otherwise only for specified user.

func (*Store) CountFactHistory added in v0.10.0 

func (s *Store) CountFactHistory() (int64, error)

CountFactHistory returns the total number of fact_history records.

func (*Store) CountFactsOnOrphanedTopics added in v0.10.0 

func (s *Store) CountFactsOnOrphanedTopics(userID ScopeID) (int, error)

CountFactsOnOrphanedTopics counts facts linked to orphaned topics. If userID is 0, counts for all users.

func (*Store) CountOrphanedTopics added in v0.10.0 

func (s *Store) CountOrphanedTopics(userID ScopeID) (int, error)

CountOrphanedTopics counts topics with no messages linked to them. If userID is 0, counts for all users.

func (*Store) CountOverlappingTopics added in v0.10.0 

func (s *Store) CountOverlappingTopics(userID ScopeID) (int, error)

CountOverlappingTopics counts pairs of topics with overlapping message ranges. If userID is 0, counts for all users.

func (*Store) CountPeopleWithoutEmbedding added in v0.10.0 

func (s *Store) CountPeopleWithoutEmbedding(userID ScopeID) (int, error)

CountPeopleWithoutEmbedding returns count of people missing embeddings.

func (*Store) CreateOutboundDelivery added in v0.11.0 

func (s *Store) CreateOutboundDelivery(delivery OutboundDelivery, operations []OutboundDeliveryOperation) (int64, error)

func (*Store) CreateTopic added in v0.10.0 

func (s *Store) CreateTopic(topic Topic) (int64, error)

func (*Store) DeleteAllFacts added in v0.10.0 

func (s *Store) DeleteAllFacts(userID ScopeID) error

DeleteAllFacts removes all facts for a user in a single query.

func (*Store) DeleteAllPeople added in v0.10.0 

func (s *Store) DeleteAllPeople(userID ScopeID) error

DeleteAllPeople removes all people for a user in a single query.

func (*Store) DeleteAllTopics added in v0.10.0 

func (s *Store) DeleteAllTopics(userID ScopeID) error

DeleteAllTopics removes all topics for a user in a single query.

func (*Store) DeleteFact added in v0.10.0 

func (s *Store) DeleteFact(userID ScopeID, id int64) error

func (*Store) DeletePerson added in v0.10.0 

func (s *Store) DeletePerson(userID ScopeID, personID int64) error

DeletePerson removes a person record.

func (*Store) DeleteTopic added in v0.10.0 

func (s *Store) DeleteTopic(userID ScopeID, id int64) error

func (*Store) DeleteTopicCascade added in v0.10.0 

func (s *Store) DeleteTopicCascade(userID ScopeID, id int64) error

func (*Store) FindPersonByAlias added in v0.10.0 

func (s *Store) FindPersonByAlias(userID ScopeID, alias string) ([]Person, error)

FindPersonByAlias finds people whose aliases contain the given string. Returns multiple matches since aliases might overlap.

func (*Store) FindPersonByExternalID added in v0.10.0 

func (s *Store) FindPersonByExternalID(userID ScopeID, transport, nativeID string) (*Person, error)

FindPersonByExternalID finds a person by their transport-neutral external id (transport, native_id). Introduced in v0.10; existing Telegram people are backfilled to ('telegram', telegram_id) by migration 011.

func (*Store) FindPersonByName added in v0.10.0 

func (s *Store) FindPersonByName(userID ScopeID, name string) (*Person, error)

FindPersonByName finds a person by their display name (exact match).

func (*Store) FindPersonByTelegramID added in v0.10.0 

func (s *Store) FindPersonByTelegramID(userID ScopeID, telegramID int64) (*Person, error)

FindPersonByTelegramID finds a person by their Telegram ID.

func (*Store) FindPersonByUsername added in v0.10.0 

func (s *Store) FindPersonByUsername(userID ScopeID, username string) (*Person, error)

FindPersonByUsername finds a person by their @username (without @).

func (*Store) FixContaminatedTopics added in v0.10.0 

func (s *Store) FixContaminatedTopics(userID ScopeID) (int64, error)

FixContaminatedTopics removes foreign messages from contaminated topics by setting their topic_id to NULL. Returns the number of messages unlinked. If userID is 0, fixes all; otherwise only topics owned by specified user.

func (*Store) GetAgentLogFull added in v0.10.0 

func (s *Store) GetAgentLogFull(ctx context.Context, id int64, userID ScopeID) (*AgentLog, error)

GetAgentLogFull fetches a single agent log with full context data. Validates userID to prevent cross-user access (security).

func (*Store) GetAgentLogs added in v0.10.0 

func (s *Store) GetAgentLogs(agentType string, userID ScopeID, limit int) ([]AgentLog, error)

GetAgentLogs returns the most recent agent logs for a specific agent type. If userID is 0, returns logs for all users.

func (*Store) GetAgentLogsExtended added in v0.10.0 

func (s *Store) GetAgentLogsExtended(filter AgentLogFilter, limit, offset int) (AgentLogResult, error)

GetAgentLogsExtended returns agent logs with filtering and pagination.

func (*Store) GetAllFacts added in v0.10.0 

func (s *Store) GetAllFacts() ([]Fact, error)

GetAllFacts retrieves all facts across all users. WARNING: Cross-user access - used for vector index loading only.

func (*Store) GetAllPeople added in v0.10.0 

func (s *Store) GetAllPeople() ([]Person, error)

GetAllPeople retrieves all people across all users. WARNING: Cross-user access - used for vector index loading only.

func (*Store) GetAllTopics added in v0.10.0 

func (s *Store) GetAllTopics() ([]Topic, error)

GetAllTopics retrieves all topics across all users. WARNING: Cross-user access - used for vector index loading and admin operations only. Caller must handle data isolation if needed.

func (*Store) GetAllUsers added in v0.10.0 

func (s *Store) GetAllUsers() ([]User, error)

func (*Store) GetArtifact added in v0.10.0 

func (s *Store) GetArtifact(userID ScopeID, artifactID int64) (*Artifact, error)

GetArtifact retrieves an artifact by ID and user ID.

func (*Store) GetArtifacts added in v0.10.0 

func (s *Store) GetArtifacts(filter ArtifactFilter, limit, offset int) ([]Artifact, int64, error)

GetArtifacts retrieves artifacts for a user with optional filters and pagination. UserID is REQUIRED for data isolation.

func (*Store) GetArtifactsByIDs added in v0.10.0 

func (s *Store) GetArtifactsByIDs(userID ScopeID, artifactIDs []int64) ([]Artifact, error)

GetArtifactsByIDs retrieves artifacts by their IDs (batch load).

func (*Store) GetArtifactsNeedingReembed added in v0.10.0 

func (s *Store) GetArtifactsNeedingReembed(expectedVersion string, limit int) ([]ReembedCandidate, error)

func (*Store) GetByHash added in v0.10.0 

func (s *Store) GetByHash(userID ScopeID, contentHash string) (*Artifact, error)

GetByHash retrieves an artifact by content hash and user ID. Used for deduplication checks.

func (*Store) GetChannel added in v0.10.0 

func (s *Store) GetChannel(scopeID ScopeID) (*Channel, error)

GetChannel returns the channel for a scope id, or nil if the scope is not a channel.

func (*Store) GetContaminatedTopics added in v0.10.0 

func (s *Store) GetContaminatedTopics(userID ScopeID) ([]ContaminatedTopic, error)

GetContaminatedTopics finds topics that contain messages from users other than the topic owner. If userID is 0, checks all topics; otherwise only topics owned by userID.

func (*Store) GetDBSize added in v0.10.0 

func (s *Store) GetDBSize() (int64, error)

GetDBSize returns the size of the database in bytes. SQLite uses the file size; Postgres uses pg_database_size(current_database()).

func (*Store) GetDashboardStats added in v0.10.0 

func (s *Store) GetDashboardStats(userID ScopeID) (*DashboardStats, error)

func (*Store) GetFactHistory added in v0.10.0 

func (s *Store) GetFactHistory(userID ScopeID, limit int) ([]FactHistory, error)

func (*Store) GetFactHistoryExtended added in v0.10.0 

func (s *Store) GetFactHistoryExtended(filter FactHistoryFilter, limit, offset int, sortBy, sortDir string) (FactHistoryResult, error)

func (*Store) GetFactStats added in v0.10.0 

func (s *Store) GetFactStats() (FactStats, error)

func (*Store) GetFactStatsByUser added in v0.10.0 

func (s *Store) GetFactStatsByUser(userID ScopeID) (FactStats, error)

func (*Store) GetFacts added in v0.10.0 

func (s *Store) GetFacts(userID ScopeID) ([]Fact, error)

func (*Store) GetFactsAfterID added in v0.10.0 

func (s *Store) GetFactsAfterID(minID int64) ([]Fact, error)

GetFactsAfterID retrieves facts created after given ID across all users. WARNING: Cross-user access - used for incremental vector index updates.

func (*Store) GetFactsByIDs added in v0.10.0 

func (s *Store) GetFactsByIDs(userID ScopeID, ids []int64) ([]Fact, error)

func (*Store) GetFactsByTopicID added in v0.10.0 

func (s *Store) GetFactsByTopicID(userID ScopeID, topicID int64) ([]Fact, error)

func (*Store) GetFactsNeedingReembed added in v0.10.0 

func (s *Store) GetFactsNeedingReembed(expectedVersion string, limit int) ([]ReembedCandidate, error)

func (*Store) GetFlags added in v0.10.2 

func (s *Store) GetFlags(userID ScopeID, limit int) ([]Flag, error)

GetFlags returns the most recent flags for a user, newest first. If userID is empty, returns flags across all users (operator/debug listing).

func (*Store) GetHistoryArtifactReferences added in v0.11.0 

func (s *Store) GetHistoryArtifactReferences(userID ScopeID, historyID int64) ([]HistoryArtifactReference, error)

GetHistoryArtifactReferences returns the provenance-preserving associations for one logical reply. Ordering is explicit and independent of insert order.

func (*Store) GetIdentity added in v0.10.0 

func (s *Store) GetIdentity(transport, nativeID string) (*Identity, error)

GetIdentity returns the identity row for (transport, nativeID), or nil if the handle has never been mapped to a scope.

func (*Store) GetLastTopicEndMessageID added in v0.10.0 

func (s *Store) GetLastTopicEndMessageID(userID ScopeID) (int64, error)

func (*Store) GetMemoryBank added in v0.10.0 

func (s *Store) GetMemoryBank(userID ScopeID) (string, error)

func (*Store) GetMergeCandidates added in v0.10.0 

func (s *Store) GetMergeCandidates(userID ScopeID) ([]MergeCandidate, error)

func (*Store) GetMessagesByIDs added in v0.10.0 

func (s *Store) GetMessagesByIDs(userID ScopeID, ids []int64) ([]Message, error)

func (*Store) GetMessagesByTopicID added in v0.10.0 

func (s *Store) GetMessagesByTopicID(ctx context.Context, topicID int64) ([]Message, error)

func (*Store) GetMessagesInRange added in v0.10.0 

func (s *Store) GetMessagesInRange(ctx context.Context, userID ScopeID, startID, endID int64) ([]Message, error)

func (*Store) GetOrCreateChannel added in v0.10.0 

func (s *Store) GetOrCreateChannel(transport, nativeID, displayName string) (ScopeID, error)

GetOrCreateChannel returns the channel scope id for (transport, nativeID), creating the scopes + channels rows on first sight. The id is deterministic (PassthroughScopeID), so it matches what ResolveScope produces for a channel. displayName is refreshed when non-empty and never clobbered with "".

func (*Store) GetOrCreatePrincipal added in v0.10.0 

func (s *Store) GetOrCreatePrincipal(in PrincipalInput) (ScopeID, bool, error)

GetOrCreatePrincipal returns the scope id for the person described by in, creating a fresh principal scope if none matches. Dedup order: object_guid (when present) then ad_login. On a match, missing attributes are backfilled (notably object_guid arriving later) without clobbering existing values with empties. The bool reports whether a new principal was created.

func (*Store) GetOrphanedTopicIDs added in v0.10.0 

func (s *Store) GetOrphanedTopicIDs(userID ScopeID) ([]int64, error)

GetOrphanedTopicIDs returns IDs of topics with no messages linked. If userID is 0, returns for all users.

func (*Store) GetOutboundDelivery added in v0.11.0 

func (s *Store) GetOutboundDelivery(deliveryID int64) (*OutboundDelivery, []OutboundDeliveryOperation, error)

func (*Store) GetOutboundDeliveryByTransportMessage added in v0.11.0 

func (s *Store) GetOutboundDeliveryByTransportMessage(userID ScopeID, transport, conversationID, transportMsgID string) (*OutboundDelivery, error)

GetOutboundDeliveryByTransportMessage resolves a stable ID recorded for a confirmed operation even when the logical reply has not acquired a history row (for example a confirmed prefix followed by an unknown operation, or the crash window immediately after final delivery confirmation). The query is content-free and exact across scope, transport, conversation, and message.

func (*Store) GetOverlappingTopics added in v0.10.0 

func (s *Store) GetOverlappingTopics(userID ScopeID) ([]OverlappingPair, error)

GetOverlappingTopics returns pairs of topics with overlapping message ranges. If userID is 0, returns for all users.

func (*Store) GetPendingArtifacts added in v0.10.0 

func (s *Store) GetPendingArtifacts(userID ScopeID, maxRetries int) ([]Artifact, error)

GetPendingArtifacts retrieves artifacts ready for processing. Includes: - state='pending' (new artifacts) - state='failed' with retry_count < maxRetries and sufficient backoff elapsed (v0.6.0 - CRIT-3) Backoff schedule: 1 min (retry 0), 5 min (retry 1), 30 min (retry 2+)

func (*Store) GetPeople added in v0.10.0 

func (s *Store) GetPeople(userID ScopeID) ([]Person, error)

GetPeople retrieves all people for a user.

func (*Store) GetPeopleAfterID added in v0.10.0 

func (s *Store) GetPeopleAfterID(minID int64) ([]Person, error)

GetPeopleAfterID retrieves people created after a given ID across all users. WARNING: Cross-user access - used for incremental vector index updates.

func (*Store) GetPeopleByIDs added in v0.10.0 

func (s *Store) GetPeopleByIDs(userID ScopeID, ids []int64) ([]Person, error)

GetPeopleByIDs retrieves people by their IDs.

func (*Store) GetPeopleExtended added in v0.10.0 

func (s *Store) GetPeopleExtended(filter PersonFilter, limit, offset int, sortBy, sortDir string) (PersonResult, error)

GetPeopleExtended retrieves people with filtering and pagination.

func (*Store) GetPeopleNeedingReembed added in v0.10.0 

func (s *Store) GetPeopleNeedingReembed(expectedVersion string, limit int) ([]ReembedCandidate, error)

GetPeopleNeedingReembed composes the search text via the canonical ComposePersonEmbeddingText, the SAME function the live write path uses, so a re-embed reproduces existing vectors instead of drifting them. It MUST pull every field that composer reads (display_name, username, aliases, bio) — a missing column here silently produces a different text and a drifted vector.

func (*Store) GetPeopleWithoutEmbedding added in v0.10.0 

func (s *Store) GetPeopleWithoutEmbedding(userID ScopeID) ([]Person, error)

GetPeopleWithoutEmbedding returns people missing embeddings.

func (*Store) GetPerson added in v0.10.0 

func (s *Store) GetPerson(userID ScopeID, personID int64) (*Person, error)

GetPerson retrieves a single person by ID.

func (*Store) GetPrincipal added in v0.10.0 

func (s *Store) GetPrincipal(scopeID ScopeID) (*Principal, error)

GetPrincipal returns the principal for a scope id, or nil if the scope is not a principal.

func (*Store) GetPrivacyMode added in v0.11.0 

func (s *Store) GetPrivacyMode(userID ScopeID) (bool, error)

GetPrivacyMode reports whether the scope's do-not-store mode is on. A missing users row means off.

func (*Store) GetRecentHistory added in v0.10.0 

func (s *Store) GetRecentHistory(userID ScopeID, limit int) ([]Message, error)

func (*Store) GetRecentSessionMessages added in v0.10.0 

func (s *Store) GetRecentSessionMessages(ctx context.Context, userID ScopeID, limit int, excludeIDs []int64) ([]Message, error)

GetRecentSessionMessages returns the last N unprocessed messages (topic_id IS NULL) for artifact context (v0.6.0). Excludes messageIDs to avoid duplicates with MessageGroup messages.

func (*Store) GetReplyByTransportID added in v0.10.2 

func (s *Store) GetReplyByTransportID(userID ScopeID, transportMsgID string) (*Message, error)

GetReplyByTransportID returns the assistant reply for a given transport-native message id, or (nil, nil) if there is no match for this user. Used by the inbound-reaction handler to resolve a reacted message to its trace; a miss means the reaction was on a non-bot / unindexed message and is ignored.

func (*Store) GetReplyByTransportMessage added in v0.11.0 

func (s *Store) GetReplyByTransportMessage(userID ScopeID, transport, conversationID, transportMsgID string) (*Message, error)

func (*Store) GetSessionArtifacts added in v0.10.0 

func (s *Store) GetSessionArtifacts(ctx context.Context, userID ScopeID, limit int, maxAge time.Duration) ([]Artifact, error)

GetSessionArtifacts returns artifacts attached to messages still in the active session (history rows with topic_id IS NULL). Used to ensure freshly-created files are exposed to the reranker even when their summary embedding doesn't match the next user query.

Filters:

  • state IN ('ready', 'pending', 'processing') — a just-sent file is 'pending' until the Extractor picks it up, and that window is exactly when the user asks about it; the file itself is loadable regardless, only the summary is missing. 'failed' stays excluded so poisoned artifacts (e.g. safety-blocked ones) are never re-surfaced.
  • message_id > 0 (skip in-flight rows where assistant-side message_id assignment hasn't completed)
  • created_at within maxAge window (safety cap for stalled sessions)

Double user_id filter (a.user_id AND h.user_id) is intentional defense-in-depth per the project's user-isolation invariants — session-aware queries with JOIN must enforce isolation on every joined table.

func (*Store) GetStats added in v0.10.0 

func (s *Store) GetStats() (map[ScopeID]Stat, error)

func (*Store) GetTableSizes added in v0.10.0 

func (s *Store) GetTableSizes() ([]TableSize, error)

GetTableSizes returns the size of each table in bytes. SQLite uses the dbstat virtual table; Postgres uses pg_total_relation_size over public tables.

func (*Store) GetTopics added in v0.10.0 

func (s *Store) GetTopics(userID ScopeID) ([]Topic, error)

func (*Store) GetTopicsAfterID added in v0.10.0 

func (s *Store) GetTopicsAfterID(minID int64) ([]Topic, error)

GetTopicsAfterID retrieves topics created after given ID across all users. WARNING: Cross-user access - used for incremental vector index updates. Caller must handle data isolation if needed.

func (*Store) GetTopicsByIDs added in v0.10.0 

func (s *Store) GetTopicsByIDs(userID ScopeID, ids []int64) ([]Topic, error)

func (*Store) GetTopicsExtended added in v0.10.0 

func (s *Store) GetTopicsExtended(filter TopicFilter, limit, offset int, sortBy, sortDir string) (TopicResult, error)

func (*Store) GetTopicsNeedingReembed added in v0.10.0 

func (s *Store) GetTopicsNeedingReembed(expectedVersion string, limit int) ([]ReembedCandidate, error)

GetTopicsNeedingReembed returns topics whose embedding_version is not the expected one. Content is the topic summary.

func (*Store) GetTopicsPendingFacts added in v0.10.0 

func (s *Store) GetTopicsPendingFacts(userID ScopeID) ([]Topic, error)

func (*Store) GetUnprocessedMessages added in v0.10.0 

func (s *Store) GetUnprocessedMessages(userID ScopeID) ([]Message, error)

func (*Store) ImportMessage added in v0.10.0 

func (s *Store) ImportMessage(userID ScopeID, message Message) error

func (*Store) IncrementContextLoadCount added in v0.10.0 

func (s *Store) IncrementContextLoadCount(userID ScopeID, artifactIDs []int64) error

IncrementContextLoadCount increments the load counter for artifacts and updates last_loaded_at timestamp. Called asynchronously after artifacts are successfully loaded into LLM context (v0.6.0).

func (*Store) Init added in v0.10.0 

func (s *Store) Init() error

Init creates/upgrades the schema for the active backend. SQLite runs the legacy bootstrap DDL plus the incremental migration runner; Postgres applies a single consolidated end-state schema (greenfield, no incremental replay).

func (*Store) IsChannelScope added in v0.10.0 

func (s *Store) IsChannelScope(id ScopeID) (bool, error)

IsChannelScope reports whether id names a channel scope (scope_type='channel'). A scope's type is fixed at mint time, so the result is memoized for the process lifetime. Absence of a row means a DM/person scope (Telegram passthrough writes no row; Mattermost DMs are scope_type='user'; principals are 'principal'), so it returns false.

func (*Store) LinkOutboundDeliveryHistory added in v0.11.0 

func (s *Store) LinkOutboundDeliveryHistory(userID ScopeID, deliveryID, historyID int64) error

func (*Store) LinkReplyTransportMessages added in v0.11.0 

func (s *Store) LinkReplyTransportMessages(userID ScopeID, historyID int64, messages []TransportMessage) error

LinkReplyTransportMessages links a fully materialized list to an exact assistant history row. It is atomic and idempotent for the same mapping.

func (*Store) MarkInterruptedOutboundDeliveriesUnknown added in v0.11.0 

func (s *Store) MarkInterruptedOutboundDeliveriesUnknown() (int64, error)

MarkInterruptedOutboundDeliveriesUnknown seals every nonterminal delivery at process restart. An operation in the non-idempotent "sending" window becomes unknown/interrupted; an operation still planned is known not to have started and becomes skipped. This also closes the crash windows before the first Mark and between a confirmed operation and the next Mark, so no delivery remains planned/sending forever after startup recovery.

func (*Store) MarkOutboundDeliveryOperationSending added in v0.11.0 

func (s *Store) MarkOutboundDeliveryOperationSending(deliveryID int64, ordinal int) error

func (*Store) MergePeople added in v0.10.0 

func (s *Store) MergePeople(userID ScopeID, targetID, sourceID int64, newBio string, newAliases []string, newUsername *string, newTelegramID *int64) error

MergePeople merges source person into target person, then deletes source. newUsername and newTelegramID are the values to set (only if non-nil/non-zero). If target already has username/telegram_id, those are preserved (callers should decide which to keep).

func (*Store) PersistOutboundDeliveryReply added in v0.11.0 

func (s *Store) PersistOutboundDeliveryReply(userID ScopeID, deliveryID int64, message Message, artifactIDs []int64) (int64, error)

func (*Store) PersistOutboundDeliveryReplyWithArtifacts added in v0.11.0 

func (s *Store) PersistOutboundDeliveryReplyWithArtifacts(userID ScopeID, deliveryID int64, message Message, artifacts PersistOutboundArtifacts) (int64, error)

PersistOutboundDeliveryReplyWithArtifacts is the provenance-preserving V2 persistence path. It atomically creates the assistant history row, links the confirmed transport messages and delivery, assigns only in-flight (message_id=0) artifacts to their creator reply, and records ordered M:N references for every delivered generated or stored artifact.

func (*Store) PutIdentity added in v0.10.0 

func (s *Store) PutIdentity(transport, nativeID string, scopeID ScopeID) error

PutIdentity upserts the (transport, nativeID) → scopeID mapping. Re-pointing an existing handle to a different scope is allowed (the resolver owns that policy).

func (*Store) RecalculateTopicRanges added in v0.10.0 

func (s *Store) RecalculateTopicRanges(userID ScopeID) (int, error)

RecalculateTopicRanges recalculates start_msg_id and end_msg_id for all topics based on actual message assignments. If userID is 0, recalculates for all users.

func (*Store) RecalculateTopicSizes added in v0.10.0 

func (s *Store) RecalculateTopicSizes(userID ScopeID) (int, error)

RecalculateTopicSizes recalculates size_chars for all topics based on actual message content. If userID is 0, recalculates for all users.

func (*Store) RecoverArtifactStates added in v0.10.0 

func (s *Store) RecoverArtifactStates(threshold time.Duration) error

RecoverArtifactStates resets zombie 'processing' states to 'pending'. Called on startup to recover from crashes or interruptions. Only recovers artifacts that have been in 'processing' state for longer than threshold to avoid re-processing actively processing artifacts.

func (*Store) ResetUserData added in v0.10.0 

func (s *Store) ResetUserData(userID ScopeID) error

func (*Store) SetEmbeddingVersion added in v0.10.3 

func (s *Store) SetEmbeddingVersion(version string)

SetEmbeddingVersion declares which embedding model/dimension produced the vectors that callers pass to Add*/Update* methods from now on. Compose the value with EmbeddingVersion.

func (*Store) SetPrivacyMode added in v0.11.0 

func (s *Store) SetPrivacyMode(userID ScopeID, enabled bool) error

SetPrivacyMode toggles the scope's do-not-store mode. While enabled, the bot writes new history rows with do_not_store=1 so their content never enters long-term memory (topics, facts, embeddings). Upserts so the flag works even before the scope has a users row.

func (*Store) SetReplyTransportID added in v0.10.2 

func (s *Store) SetReplyTransportID(userID ScopeID, transportMsgID string) error

SetReplyTransportID back-fills the transport-native message id on the assistant reply just stored for a turn, after it has been sent (the id isn't known at insert time). It targets the user's most recent assistant row that is still unlinked (message_id IS NULL) — which, because message-group turns are serialized per user, is exactly the reply we just inserted. Needs no row-id read and no trace_id, so it works whether or not tracing is enabled. user_id-scoped (Critical Data Invariant). No-op if there is no unlinked reply.

func (*Store) SetTopicConsolidationChecked added in v0.10.0 

func (s *Store) SetTopicConsolidationChecked(userID ScopeID, topicID int64, checked bool) error

func (*Store) SetTopicFactsExtracted added in v0.10.0 

func (s *Store) SetTopicFactsExtracted(userID ScopeID, topicID int64, extracted bool) error

func (*Store) UpdateArtifact added in v0.10.0 

func (s *Store) UpdateArtifact(artifact Artifact) error

UpdateArtifact updates an artifact's metadata. embedding_version is re-stamped only when the vector actually changed, so state/retry bookkeeping writes that round-trip the stored embedding cannot mark an old-space vector as already migrated (SET expressions see pre-update column values on both SQLite and Postgres).

func (*Store) UpdateArtifactEmbeddingVersion added in v0.10.0 

func (s *Store) UpdateArtifactEmbeddingVersion(id int64, emb []float32, version string) error

func (*Store) UpdateFact added in v0.10.0 

func (s *Store) UpdateFact(fact Fact) error

func (*Store) UpdateFactEmbeddingVersion added in v0.10.0 

func (s *Store) UpdateFactEmbeddingVersion(id int64, emb []float32, version string) error

func (*Store) UpdateFactHistoryTopic added in v0.10.0 

func (s *Store) UpdateFactHistoryTopic(oldTopicID, newTopicID int64) error

func (*Store) UpdateFactsTopic added in v0.10.0 

func (s *Store) UpdateFactsTopic(userID ScopeID, oldTopicID, newTopicID int64) error

UpdateFactsTopic updates topic_id for all facts belonging to a user and old topic.

func (*Store) UpdateMemoryBank added in v0.10.0 

func (s *Store) UpdateMemoryBank(userID ScopeID, content string) error

func (*Store) UpdateMessageID added in v0.10.0 

func (s *Store) UpdateMessageID(userID ScopeID, artifactID, messageID int64) error

UpdateMessageID links an artifact to a history message. Called after message is saved to history (message_id is not known during file processing). Requires userID for proper data isolation (CRIT-2 security fix).

func (*Store) UpdateMessageTopic added in v0.10.0 

func (s *Store) UpdateMessageTopic(userID ScopeID, messageID, topicID int64) error

func (*Store) UpdateMessagesTopicInRange added in v0.10.0 

func (s *Store) UpdateMessagesTopicInRange(ctx context.Context, userID ScopeID, startMsgID, endMsgID, topicID int64) error

func (*Store) UpdatePerson added in v0.10.0 

func (s *Store) UpdatePerson(person Person) error

UpdatePerson updates an existing person record.

func (*Store) UpdatePersonEmbeddingVersion added in v0.10.0 

func (s *Store) UpdatePersonEmbeddingVersion(id int64, emb []float32, version string) error

func (*Store) UpdateTopicEmbeddingVersion added in v0.10.0 

func (s *Store) UpdateTopicEmbeddingVersion(id int64, emb []float32, version string) error

func (*Store) UpsertUser added in v0.10.0 

func (s *Store) UpsertUser(user User) error

type TableSize added in v0.3.5 

type TableSize struct {
	Name  string
	Bytes int64
}

TableSize represents the size of a database table.

type Topic 

type Topic struct {
	ID                   int64
	UserID               ScopeID
	Summary              string
	StartMsgID           int64
	EndMsgID             int64
	SizeChars            int // Total character count of all messages in topic
	Embedding            []float32
	FactsExtracted       bool
	IsConsolidated       bool
	ConsolidationChecked bool
	CreatedAt            time.Time
}

type TopicExtended 

type TopicExtended struct {
	Topic
	FactsCount   int
	MessageCount int
}

type TopicFilter 

type TopicFilter struct {
	UserID         ScopeID
	Search         string
	HasFacts       *bool // nil = all, true = yes, false = no
	IsConsolidated *bool // nil = all
	TopicID        *int64
}

type TopicRepository 

type TopicRepository interface {
	AddTopic(topic Topic) (int64, error)
	AddTopicWithoutMessageUpdate(topic Topic) (int64, error)
	CreateTopic(topic Topic) (int64, error)
	DeleteTopic(userID ScopeID, id int64) error
	DeleteTopicCascade(userID ScopeID, id int64) error
	GetLastTopicEndMessageID(userID ScopeID) (int64, error)
	GetAllTopics() ([]Topic, error)
	GetTopicsAfterID(minID int64) ([]Topic, error)
	GetTopicsByIDs(userID ScopeID, ids []int64) ([]Topic, error)
	GetTopics(userID ScopeID) ([]Topic, error)
	SetTopicFactsExtracted(userID ScopeID, topicID int64, extracted bool) error
	SetTopicConsolidationChecked(userID ScopeID, topicID int64, checked bool) error
	GetTopicsPendingFacts(userID ScopeID) ([]Topic, error)
	GetTopicsExtended(filter TopicFilter, limit, offset int, sortBy, sortDir string) (TopicResult, error)
	GetMergeCandidates(userID ScopeID) ([]MergeCandidate, error)

	// v0.7.0: embedding migration
	GetTopicsNeedingReembed(expectedVersion string, limit int) ([]ReembedCandidate, error)
	UpdateTopicEmbeddingVersion(id int64, emb []float32, version string) error
}

TopicRepository handles topic operations.

Topics are compressed summaries of conversation chunks created after session archival (inactivity timeout or force-close). Each topic has an embedding vector for RAG retrieval.

Key Fields:

  • StartMsgID/EndMsgID: Message range (inclusive)
  • SizeChars: Total character count for size tracking
  • Embedding: Vector for semantic search

Note: Message IDs within a topic are guaranteed to be from the same user.

type TopicResult 

type TopicResult struct {
	Data       []TopicExtended
	TotalCount int
}

type TransportMessage added in v0.11.0 

type TransportMessage struct {
	Transport      string
	ConversationID string
	MessageID      string
	Ordinal        int
	IsPrimary      bool
}

TransportMessage is one persistent transport message belonging to a logical assistant reply. ConversationID is mandatory because native message ids are not necessarily globally unique (notably outside Telegram private chats).

type User 

type User struct {
	ID        ScopeID
	Username  string
	FirstName string
	LastName  string
	LastSeen  time.Time
}

type UserRepository 

type UserRepository interface {
	UpsertUser(user User) error
	GetAllUsers() ([]User, error)
	ResetUserData(userID ScopeID) error
	// SetPrivacyMode toggles the scope's do-not-store mode (migration 018);
	// GetPrivacyMode reads it, a missing users row meaning off.
	SetPrivacyMode(userID ScopeID, enabled bool) error
	GetPrivacyMode(userID ScopeID) (bool, error)
}

UserRepository handles user data operations.

Source Files

View all Source files

Directories

Path Synopsis
Package migrations handles database schema migrations with version tracking.
 Package migrations handles database schema migrations with version tracking.

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.