postgres

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Apr 8, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Index

Constants

View Source 
const DefaultOAuthStateTTL = 10 * time.Minute

DefaultOAuthStateTTL is the default time-to-live for OAuth states.

Variables

View Source 
var (
	// ErrInvalidKeySize is returned when the encryption key is not 32 bytes.
	ErrInvalidKeySize = errors.New("encryption key must be 32 bytes")

	// ErrInvalidBlobSize is returned when the encrypted blob is too small.
	ErrInvalidBlobSize = errors.New("encrypted blob is too small")

	// ErrUnsupportedVersion is returned when the blob version is not supported.
	ErrUnsupportedVersion = errors.New("unsupported secret blob version")

	// ErrDecryptionFailed is returned when decryption fails (wrong key or corrupted data).
	ErrDecryptionFailed = errors.New("failed to decrypt secret blob")
)

Functions

func NullString 

func NullString(s *string) sql.NullString

NullString converts a string pointer to sql.NullString

func NullTime 

func NullTime(t *time.Time) sql.NullTime

NullTime converts a time pointer to sql.NullTime

func StringPtr 

func StringPtr(ns sql.NullString) *string

StringPtr converts sql.NullString to string pointer

func TimePtr 

func TimePtr(nt sql.NullTime) *time.Time

TimePtr converts sql.NullTime to time pointer

Types

type AdvisoryLock 

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

AdvisoryLock implements DistributedLock using PostgreSQL advisory locks.

IMPORTANT LIMITATIONS: - Advisory locks are connection-scoped, not TTL-based - If the connection is lost, the lock is automatically released - TTL parameter is ignored (locks don't expire automatically) - Extend is a no-op since locks don't have TTL

For production multi-worker deployments, Redis locks are recommended. This is provided as a fallback when Redis is unavailable.

func NewAdvisoryLock 

func NewAdvisoryLock(db *DB) *AdvisoryLock

NewAdvisoryLock creates a new PostgreSQL advisory lock adapter.

func (*AdvisoryLock) Acquire 

func (l *AdvisoryLock) Acquire(ctx context.Context, name string, ttl time.Duration) (bool, error)

Acquire attempts to acquire a named advisory lock. Uses pg_try_advisory_lock which returns immediately without blocking.

Note: The TTL parameter is ignored - PostgreSQL advisory locks don't have TTL. The lock is held until explicitly released or the connection closes.

func (*AdvisoryLock) Extend 

func (l *AdvisoryLock) Extend(ctx context.Context, name string, ttl time.Duration) error

Extend is a no-op for PostgreSQL advisory locks since they don't have TTL. Advisory locks are held until explicitly released or the connection closes.

func (*AdvisoryLock) Ping 

func (l *AdvisoryLock) Ping(ctx context.Context) error

Ping checks if the PostgreSQL backend is healthy.

func (*AdvisoryLock) Release 

func (l *AdvisoryLock) Release(ctx context.Context, name string) error

Release releases a named advisory lock. Uses pg_advisory_unlock to release the lock. Safe to call even if the lock is not held (returns false but no error).

type CapabilityStore 

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

CapabilityStore implements driven.CapabilityStore using PostgreSQL

func NewCapabilityStore 

func NewCapabilityStore(db *DB) *CapabilityStore

NewCapabilityStore creates a new CapabilityStore

func (*CapabilityStore) GetPreferences 

func (s *CapabilityStore) GetPreferences(ctx context.Context, teamID string) (*domain.CapabilityPreferences, error)

GetPreferences retrieves capability preferences for a team

func (*CapabilityStore) SavePreferences 

func (s *CapabilityStore) SavePreferences(ctx context.Context, prefs *domain.CapabilityPreferences) error

SavePreferences persists capability preferences for a team

type ChunkStore 

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

ChunkStore implements driven.ChunkStore using PostgreSQL Note: Embeddings are stored in the vector store (pgvector), not here

func NewChunkStore 

func NewChunkStore(db *DB) *ChunkStore

NewChunkStore creates a new ChunkStore

func (*ChunkStore) Delete 

func (s *ChunkStore) Delete(ctx context.Context, id string) error

Delete deletes a chunk

func (*ChunkStore) DeleteBatch 

func (s *ChunkStore) DeleteBatch(ctx context.Context, ids []string) error

DeleteBatch deletes multiple chunks by ID

func (*ChunkStore) DeleteByDocument 

func (s *ChunkStore) DeleteByDocument(ctx context.Context, documentID string) error

DeleteByDocument deletes all chunks for a document

func (*ChunkStore) DeleteBySource 

func (s *ChunkStore) DeleteBySource(ctx context.Context, sourceID string) error

DeleteBySource deletes all chunks for a source

func (*ChunkStore) GetByDocument 

func (s *ChunkStore) GetByDocument(ctx context.Context, documentID string) ([]*domain.Chunk, error)

GetByDocument retrieves all chunks for a document

func (*ChunkStore) GetChunkIDs 

func (s *ChunkStore) GetChunkIDs(ctx context.Context, documentID string) ([]string, error)

GetChunkIDs returns all chunk IDs for a document (useful for search index cleanup)

func (*ChunkStore) GetChunkIDsBySource 

func (s *ChunkStore) GetChunkIDsBySource(ctx context.Context, sourceID string) ([]string, error)

GetChunkIDsBySource returns all chunk IDs for a source (useful for search index cleanup)

func (*ChunkStore) Save 

func (s *ChunkStore) Save(ctx context.Context, chunk *domain.Chunk) error

Save creates or updates a chunk

func (*ChunkStore) SaveBatch 

func (s *ChunkStore) SaveBatch(ctx context.Context, chunks []*domain.Chunk) error

SaveBatch saves multiple chunks in a transaction

type Config 

type Config struct {
	// URL is the full connection string (postgres://user:pass@host:port/db?sslmode=disable)
	URL string

	// MaxOpenConns is the maximum number of open connections
	MaxOpenConns int

	// MaxIdleConns is the maximum number of idle connections
	MaxIdleConns int

	// ConnMaxLifetime is the maximum lifetime of a connection
	ConnMaxLifetime time.Duration

	// ConnMaxIdleTime is the maximum idle time of a connection
	ConnMaxIdleTime time.Duration
}

Config holds database connection configuration

func DefaultConfig 

func DefaultConfig(url string) Config

DefaultConfig returns sensible defaults

type ConnectionStore 

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

ConnectionStore implements driven.ConnectionStore using PostgreSQL.

func NewConnectionStore 

func NewConnectionStore(db *sql.DB, encryptor *SecretEncryptor) *ConnectionStore

NewConnectionStore creates a new PostgreSQL-backed connection store.

func (*ConnectionStore) Delete 

func (s *ConnectionStore) Delete(ctx context.Context, id string) error

Delete removes a connection by ID.

func (*ConnectionStore) Get 

func (s *ConnectionStore) Get(ctx context.Context, id string) (*domain.Connection, error)

Get retrieves a connection by ID with decrypted secrets.

func (*ConnectionStore) GetByAccountID 

func (s *ConnectionStore) GetByAccountID(ctx context.Context, providerType domain.ProviderType, accountID string) (*domain.Connection, error)

GetByAccountID retrieves a connection by provider type and account ID.

func (*ConnectionStore) GetByProvider 

func (s *ConnectionStore) GetByProvider(ctx context.Context, providerType domain.ProviderType) ([]*domain.ConnectionSummary, error)

GetByProvider retrieves connections for a provider type (no secrets).

func (*ConnectionStore) List 

func (s *ConnectionStore) List(ctx context.Context) ([]*domain.ConnectionSummary, error)

List retrieves all connections as summaries (no secrets).

func (*ConnectionStore) Save 

func (s *ConnectionStore) Save(ctx context.Context, conn *domain.Connection) error

Save stores a new connection or updates an existing one.

func (*ConnectionStore) UpdateLastUsed 

func (s *ConnectionStore) UpdateLastUsed(ctx context.Context, id string) error

UpdateLastUsed updates the last_used_at timestamp.

func (*ConnectionStore) UpdateSecrets 

func (s *ConnectionStore) UpdateSecrets(ctx context.Context, id string, secrets *domain.ConnectionSecrets, expiry *time.Time) error

UpdateSecrets updates the encrypted secrets and OAuth metadata.

type DB 

type DB struct {
	*sql.DB
}

DB wraps a sql.DB connection pool with Sercha-specific functionality

func Connect 

func Connect(ctx context.Context, cfg Config) (*DB, error)

Connect establishes a database connection and runs migrations

func (*DB) Close 

func (db *DB) Close() error

Close closes the database connection

func (*DB) InitSchema 

func (db *DB) InitSchema(ctx context.Context) error

InitSchema runs the schema initialization with advisory lock protection. This is idempotent - safe to run multiple times. Uses PostgreSQL advisory locks to prevent race conditions when multiple instances start simultaneously.

func (*DB) Ping 

func (db *DB) Ping(ctx context.Context) error

Ping checks if the database is reachable

func (*DB) Transaction 

func (db *DB) Transaction(ctx context.Context, fn func(*sql.Tx) error) error

Transaction executes a function within a database transaction

type DocumentStore 

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

DocumentStore implements driven.DocumentStore using PostgreSQL

func NewDocumentStore 

func NewDocumentStore(db *DB) *DocumentStore

NewDocumentStore creates a new DocumentStore

func (*DocumentStore) Count 

func (s *DocumentStore) Count(ctx context.Context) (int, error)

Count returns total document count

func (*DocumentStore) CountBySource 

func (s *DocumentStore) CountBySource(ctx context.Context, sourceID string) (int, error)

CountBySource returns document count for a source

func (*DocumentStore) Delete 

func (s *DocumentStore) Delete(ctx context.Context, id string) error

Delete deletes a document

func (*DocumentStore) DeleteBatch 

func (s *DocumentStore) DeleteBatch(ctx context.Context, ids []string) error

DeleteBatch deletes multiple documents by ID

func (*DocumentStore) DeleteBySource 

func (s *DocumentStore) DeleteBySource(ctx context.Context, sourceID string) error

DeleteBySource deletes all documents for a source

func (*DocumentStore) Get 

func (s *DocumentStore) Get(ctx context.Context, id string) (*domain.Document, error)

Get retrieves a document by ID

func (*DocumentStore) GetByExternalID 

func (s *DocumentStore) GetByExternalID(ctx context.Context, sourceID, externalID string) (*domain.Document, error)

GetByExternalID retrieves a document by source and external ID

func (*DocumentStore) GetBySource 

func (s *DocumentStore) GetBySource(ctx context.Context, sourceID string, limit, offset int) ([]*domain.Document, error)

GetBySource retrieves all documents for a source with pagination

func (*DocumentStore) ListExternalIDs 

func (s *DocumentStore) ListExternalIDs(ctx context.Context, sourceID string) ([]string, error)

ListExternalIDs returns all external IDs for a source (for diff sync)

func (*DocumentStore) Save 

func (s *DocumentStore) Save(ctx context.Context, doc *domain.Document) error

Save creates or updates a document

func (*DocumentStore) SaveBatch 

func (s *DocumentStore) SaveBatch(ctx context.Context, docs []*domain.Document) error

SaveBatch saves multiple documents in a transaction

type OAuthStateStore 

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

OAuthStateStore implements driven.OAuthStateStore using PostgreSQL.

func NewOAuthStateStore 

func NewOAuthStateStore(db *sql.DB) *OAuthStateStore

NewOAuthStateStore creates a new PostgreSQL-backed OAuth state store.

func NewOAuthStateStoreWithTTL 

func NewOAuthStateStoreWithTTL(db *sql.DB, ttl time.Duration) *OAuthStateStore

NewOAuthStateStoreWithTTL creates an OAuth state store with custom TTL.

func (*OAuthStateStore) Cleanup 

func (s *OAuthStateStore) Cleanup(ctx context.Context) error

Cleanup removes expired states.

func (*OAuthStateStore) GetAndDelete 

func (s *OAuthStateStore) GetAndDelete(ctx context.Context, state string) (*driven.OAuthState, error)

GetAndDelete atomically retrieves and deletes the state. Uses DELETE ... RETURNING for atomic single-use semantics.

func (*OAuthStateStore) Save 

func (s *OAuthStateStore) Save(ctx context.Context, state *driven.OAuthState) error

Save stores a new OAuth state.

type SchedulerStore 

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

SchedulerStore implements driven.SchedulerStore using PostgreSQL

func NewSchedulerStore 

func NewSchedulerStore(db *DB) *SchedulerStore

NewSchedulerStore creates a new SchedulerStore

func (*SchedulerStore) DeleteScheduledTask 

func (s *SchedulerStore) DeleteScheduledTask(ctx context.Context, id string) error

DeleteScheduledTask removes a scheduled task

func (*SchedulerStore) GetDueScheduledTasks 

func (s *SchedulerStore) GetDueScheduledTasks(ctx context.Context) ([]*domain.ScheduledTask, error)

GetDueScheduledTasks retrieves scheduled tasks that are due to run

func (*SchedulerStore) GetScheduledTask 

func (s *SchedulerStore) GetScheduledTask(ctx context.Context, id string) (*domain.ScheduledTask, error)

GetScheduledTask retrieves a scheduled task by ID

func (*SchedulerStore) ListScheduledTasks 

func (s *SchedulerStore) ListScheduledTasks(ctx context.Context, teamID string) ([]*domain.ScheduledTask, error)

ListScheduledTasks retrieves all scheduled tasks for a team

func (*SchedulerStore) SaveScheduledTask 

func (s *SchedulerStore) SaveScheduledTask(ctx context.Context, task *domain.ScheduledTask) error

SaveScheduledTask creates or updates a scheduled task

func (*SchedulerStore) UpdateLastRun 

func (s *SchedulerStore) UpdateLastRun(ctx context.Context, id string, lastError string) error

UpdateLastRun updates the last run time and next run time

type SearchQueryRepository 

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

SearchQueryRepository implements driven.SearchQueryRepository using PostgreSQL

func NewSearchQueryRepository 

func NewSearchQueryRepository(db *DB) *SearchQueryRepository

NewSearchQueryRepository creates a new SearchQueryRepository

func (*SearchQueryRepository) GetSearchAnalytics 

func (r *SearchQueryRepository) GetSearchAnalytics(ctx context.Context, teamID string, period domain.AnalyticsPeriod) (*domain.SearchAnalytics, error)

GetSearchAnalytics computes aggregated search analytics for a time period

func (*SearchQueryRepository) GetSearchHistory 

func (r *SearchQueryRepository) GetSearchHistory(ctx context.Context, teamID string, limit int) ([]*domain.SearchQuery, error)

GetSearchHistory retrieves recent search queries

func (*SearchQueryRepository) GetSearchMetrics 

func (r *SearchQueryRepository) GetSearchMetrics(ctx context.Context, teamID string, period domain.AnalyticsPeriod) (*domain.SearchMetrics, error)

GetSearchMetrics computes performance metrics for a time period

func (*SearchQueryRepository) Save 

func (r *SearchQueryRepository) Save(ctx context.Context, query *domain.SearchQuery) error

Save logs a search query for analytics tracking

type SecretEncryptor 

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

SecretEncryptor handles AES-256-GCM encryption/decryption of secrets. The encrypted format is: version(1) || nonce(12) || ciphertext(N)

func NewSecretEncryptor 

func NewSecretEncryptor(key []byte) (*SecretEncryptor, error)

NewSecretEncryptor creates a new encryptor with the given 32-byte key.

func (*SecretEncryptor) Decrypt 

func (e *SecretEncryptor) Decrypt(blob []byte, value any) error

Decrypt decrypts a blob and unmarshals the result into the given value. The value should be a pointer to the target type.

func (*SecretEncryptor) DecryptString 

func (e *SecretEncryptor) DecryptString(blob []byte) (string, error)

DecryptString decrypts a blob to a string.

func (*SecretEncryptor) Encrypt 

func (e *SecretEncryptor) Encrypt(value any) ([]byte, error)

Encrypt encrypts the given value to a blob. The value is JSON-marshaled before encryption. Format: version(1) || nonce(12) || ciphertext

func (*SecretEncryptor) EncryptString 

func (e *SecretEncryptor) EncryptString(s string) ([]byte, error)

EncryptString encrypts a simple string value.

type SessionStore 

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

SessionStore implements driven.SessionStore using PostgreSQL

func NewSessionStore 

func NewSessionStore(db *DB) *SessionStore

NewSessionStore creates a new SessionStore

func (*SessionStore) Delete 

func (s *SessionStore) Delete(ctx context.Context, id string) error

Delete deletes a session

func (*SessionStore) DeleteByToken 

func (s *SessionStore) DeleteByToken(ctx context.Context, token string) error

DeleteByToken deletes a session by token

func (*SessionStore) DeleteByUser 

func (s *SessionStore) DeleteByUser(ctx context.Context, userID string) error

DeleteByUser deletes all sessions for a user (logout everywhere)

func (*SessionStore) Get 

func (s *SessionStore) Get(ctx context.Context, id string) (*domain.Session, error)

Get retrieves a session by ID

func (*SessionStore) GetByRefreshToken 

func (s *SessionStore) GetByRefreshToken(ctx context.Context, refreshToken string) (*domain.Session, error)

GetByRefreshToken retrieves a session by refresh token value

func (*SessionStore) GetByToken 

func (s *SessionStore) GetByToken(ctx context.Context, token string) (*domain.Session, error)

GetByToken retrieves a session by token value

func (*SessionStore) ListByUser 

func (s *SessionStore) ListByUser(ctx context.Context, userID string) ([]*domain.Session, error)

ListByUser lists all active sessions for a user

func (*SessionStore) Save 

func (s *SessionStore) Save(ctx context.Context, session *domain.Session) error

Save stores a session

type SettingsStore 

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

SettingsStore implements driven.SettingsStore using PostgreSQL

func NewSettingsStore 

func NewSettingsStore(db *DB) *SettingsStore

NewSettingsStore creates a new SettingsStore

func (*SettingsStore) GetAISettings 

func (s *SettingsStore) GetAISettings(ctx context.Context, teamID string) (*domain.AISettings, error)

GetAISettings retrieves AI-specific settings for a team Note: API keys and base URLs are NOT stored in database - they come from environment variables

func (*SettingsStore) GetSettings 

func (s *SettingsStore) GetSettings(ctx context.Context, teamID string) (*domain.Settings, error)

GetSettings retrieves settings for a team Note: AI configuration is managed via AISettings (ai_settings table), not here Note: semantic_search_enabled column is deprecated - use capability_preferences table instead

func (*SettingsStore) SaveAISettings 

func (s *SettingsStore) SaveAISettings(ctx context.Context, teamID string, settings *domain.AISettings) error

SaveAISettings persists AI-specific settings Note: Only provider and model are stored - API keys and base URLs come from environment

func (*SettingsStore) SaveSettings 

func (s *SettingsStore) SaveSettings(ctx context.Context, settings *domain.Settings) error

SaveSettings persists team settings Note: AI configuration is managed via SaveAISettings, not here Note: semantic_search_enabled column is deprecated - use capability_preferences table instead

type SourceStore 

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

SourceStore implements driven.SourceStore using PostgreSQL

func NewSourceStore 

func NewSourceStore(db *DB) *SourceStore

NewSourceStore creates a new SourceStore

func (*SourceStore) CountByConnection 

func (s *SourceStore) CountByConnection(ctx context.Context, connectionID string) (int, error)

CountByConnection returns the number of sources using a connection

func (*SourceStore) Delete 

func (s *SourceStore) Delete(ctx context.Context, id string) error

Delete deletes a source

func (*SourceStore) Get 

func (s *SourceStore) Get(ctx context.Context, id string) (*domain.Source, error)

Get retrieves a source by ID

func (*SourceStore) GetByName 

func (s *SourceStore) GetByName(ctx context.Context, name string) (*domain.Source, error)

GetByName retrieves a source by name

func (*SourceStore) List 

func (s *SourceStore) List(ctx context.Context) ([]*domain.Source, error)

List retrieves all sources

func (*SourceStore) ListByConnection 

func (s *SourceStore) ListByConnection(ctx context.Context, connectionID string) ([]*domain.Source, error)

ListByConnection returns sources using a connection

func (*SourceStore) ListEnabled 

func (s *SourceStore) ListEnabled(ctx context.Context) ([]*domain.Source, error)

ListEnabled retrieves all enabled sources

func (*SourceStore) Save 

func (s *SourceStore) Save(ctx context.Context, source *domain.Source) error

Save creates or updates a source

func (*SourceStore) SetEnabled 

func (s *SourceStore) SetEnabled(ctx context.Context, id string, enabled bool) error

SetEnabled updates the enabled status

func (*SourceStore) UpdateContainers 

func (s *SourceStore) UpdateContainers(ctx context.Context, id string, containers []domain.Container) error

UpdateContainers updates the selected containers for a source

type SyncStateStore 

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

SyncStateStore implements driven.SyncStateStore using PostgreSQL

func NewSyncStateStore 

func NewSyncStateStore(db *DB) *SyncStateStore

NewSyncStateStore creates a new SyncStateStore

func (*SyncStateStore) Delete 

func (s *SyncStateStore) Delete(ctx context.Context, sourceID string) error

Delete deletes sync state for a source

func (*SyncStateStore) Get 

func (s *SyncStateStore) Get(ctx context.Context, sourceID string) (*domain.SyncState, error)

Get retrieves sync state for a source

func (*SyncStateStore) List 

func (s *SyncStateStore) List(ctx context.Context) ([]*domain.SyncState, error)

List retrieves sync states for all sources

func (*SyncStateStore) Save 

func (s *SyncStateStore) Save(ctx context.Context, state *domain.SyncState) error

Save creates or updates sync state

func (*SyncStateStore) UpdateCursor 

func (s *SyncStateStore) UpdateCursor(ctx context.Context, sourceID string, cursor string) error

UpdateCursor updates the sync cursor

func (*SyncStateStore) UpdateStatus 

func (s *SyncStateStore) UpdateStatus(ctx context.Context, sourceID string, status domain.SyncStatus) error

UpdateStatus updates only the status field

type UserStore 

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

UserStore implements driven.UserStore using PostgreSQL

func NewUserStore 

func NewUserStore(db *DB) *UserStore

NewUserStore creates a new UserStore

func (*UserStore) Delete 

func (s *UserStore) Delete(ctx context.Context, id string) error

Delete deletes a user

func (*UserStore) Get 

func (s *UserStore) Get(ctx context.Context, id string) (*domain.User, error)

Get retrieves a user by ID

func (*UserStore) GetByEmail 

func (s *UserStore) GetByEmail(ctx context.Context, email string) (*domain.User, error)

GetByEmail retrieves a user by email

func (*UserStore) List 

func (s *UserStore) List(ctx context.Context, teamID string) ([]*domain.User, error)

List retrieves all users for a team

func (*UserStore) Save 

func (s *UserStore) Save(ctx context.Context, user *domain.User) error

Save creates or updates a user

func (*UserStore) UpdateLastLogin 

func (s *UserStore) UpdateLastLogin(ctx context.Context, id string) error

UpdateLastLogin updates the last login timestamp

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.