Help us shape the future of Go! Take the 2026 Go Developer Survey

sqlite

package
v0.31.0 Latest Latest
Warning

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

 Go to latest Published: Oct 28, 2026 License: Apache-2.0 Imports: 24 Imported by: 0

Documentation

Overview

Package sqlite implements filament.DataStore, filament.ScheduleStore, and the pipeline store on SQLite via the CGO-free modernc driver, with queries generated by sqlc from datastore/sqlite/queries/*.sql.

The dialect differs from datastore/postgres deliberately: times are unix milliseconds, JSON columns are TEXT, ids are TEXT UUIDs minted in Go, and the single-writer connection provides the serialization Postgres expresses with FOR UPDATE.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Migrate 

func Migrate(db *sql.DB) error

Migrate applies all pending migrations. It is safe to run on every process start; goose tracks applied versions in its own goose_db_version table. The provider is instance-scoped, so it never collides with the postgres store's migrator in the same process.

Types

type Store 

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

Store is a SQLite-backed filament.DataStore and filament.ScheduleStore.

func New 

func New(db *sql.DB) *Store

New wraps an already-open handle. Run Migrate before first use.

func NewMemory 

func NewMemory() *Store

NewMemory opens a private in-memory store without referential enforcement: the test double module tests exercise without seeding parent rows. Use Open(":memory:") for a strict in-memory database. Opening one cannot realistically fail, so failure panics rather than making every throwaway store two-valued.

func Open 

func Open(path string) (*Store, error)

Open opens (creating if needed) the database at path, applies pragmas and migrations, and returns the store. ":memory:" opens a private in-memory database with the same enforcement as a file.

func (*Store) ClaimDue 

func (s *Store) ClaimDue(ctx context.Context, now time.Time, limit int) ([]filament.ScheduleState, error)

ClaimDue leases up to limit enabled schedules not currently under an unexpired lease, stamping claimed_at in the same transaction. The single-writer connection keeps two schedulers from claiming the same row. The lease is released by the next SaveSchedule or expires after the TTL.

func (*Store) Close 

func (s *Store) Close() error

Close releases the underlying handle.

func (*Store) CreateConnection 

func (s *Store) CreateConnection(ctx context.Context, c filament.Connection) (filament.Connection, error)

CreateConnection stores a new connection at version 1, rejecting a duplicate ID.

func (*Store) CreateNotifier added in v0.3.1 

func (s *Store) CreateNotifier(ctx context.Context, n *ingestionv1.Notifier) (*ingestionv1.Notifier, error)

CreateNotifier adds a rule.

func (*Store) CreatePipeline 

func (s *Store) CreatePipeline(ctx context.Context, p *ingestionv1.Pipeline) (*ingestionv1.Pipeline, error)

CreatePipeline stores a new pipeline.

func (*Store) CreatePipelineVersion 

func (s *Store) CreatePipelineVersion(ctx context.Context, tenant filament.TenantID, pipelineID string, v *ingestionv1.PipelineVersion) (*ingestionv1.PipelineVersion, error)

CreatePipelineVersion appends an immutable graph version to a pipeline. The single-writer connection serializes the read-max-insert the postgres store expresses with a locking CTE.

func (*Store) CreatePipelineWithSchedule 

func (s *Store) CreatePipelineWithSchedule(ctx context.Context, p *ingestionv1.Pipeline, schedule *filament.ScheduleState) (*ingestionv1.Pipeline, error)

CreatePipelineWithSchedule stores a pipeline and optional schedule atomically.

func (*Store) CreateRun 

func (s *Store) CreateRun(ctx context.Context, r filament.RunState) error

CreateRun inserts the run or promotes a pre-created RunScheduled row; a row that has progressed past RunScheduled is left untouched and reported as a conflict.

func (*Store) DB 

func (s *Store) DB() *sql.DB

DB exposes the underlying handle for components backed by the same database.

func (*Store) DedupSeen 

func (s *Store) DedupSeen(ctx context.Context, tenant string, run filament.RunID, seq uint64) (bool, error)

DedupSeen reports whether seq has already been applied for (tenant, run), advancing the run's high-water mark on the first call to see it. This relies on the tracker processing each run's facts through one sequential pump, so seq only ever increases per run.

func (*Store) DeleteConnection 

func (s *Store) DeleteConnection(ctx context.Context, tenant filament.TenantID, id string) error

DeleteConnection soft-deletes the connection with the given ID; deleting a missing ID is a no-op. The name is freed for reuse by the partial unique index.

func (*Store) DeleteNotifier added in v0.3.1 

func (s *Store) DeleteNotifier(ctx context.Context, tenant filament.TenantID, pipelineID, id string) (*ingestionv1.Notifier, error)

DeleteNotifier marks a live rule deleted and returns it for secret cleanup.

func (*Store) DeletePipeline 

func (s *Store) DeletePipeline(ctx context.Context, tenant filament.TenantID, id string) error

DeletePipeline soft-deletes a pipeline and its notifiers, and removes schedules and pending runs. Versions and run history are kept.

func (*Store) DeleteResourceCheckpoint 

func (s *Store) DeleteResourceCheckpoint(ctx context.Context, tenant filament.TenantID, key filament.ResourceCheckpointKey) error

DeleteResourceCheckpoint resets durable progress for one route/resource.

func (*Store) DeleteRun 

func (s *Store) DeleteRun(ctx context.Context, tenant filament.TenantID, id filament.RunID) error

DeleteRun removes the run; resources, checkpoints, and dedup rows cascade. Missing is a no-op.

func (*Store) DeleteSchedule 

func (s *Store) DeleteSchedule(ctx context.Context, tenant filament.TenantID, id filament.ScheduleID) error

DeleteSchedule removes a schedule by id along with its pre-created scheduled runs, so nothing lingers as upcoming work. The runs reap goes first: deleting the schedules row SET-NULLs runs.schedule_id, after which the rows are unreachable by schedule id.

func (*Store) EnsureTenant 

func (s *Store) EnsureTenant(ctx context.Context, id filament.TenantID, name string) error

EnsureTenant creates or updates a tenant row.

func (*Store) EnsureUser 

func (s *Store) EnsureUser(ctx context.Context, tenant filament.TenantID, externalID string) (filament.UserID, error)

EnsureUser creates or touches the user row for a provider subject within a tenant and returns filament's id for it.

func (*Store) ListConnections 

func (s *Store) ListConnections(ctx context.Context, f filament.ConnectionFilter) ([]filament.Connection, int, error)

ListConnections returns one filtered, sorted page and its pre-page total.

func (*Store) ListNotifiers added in v0.3.1 

func (s *Store) ListNotifiers(ctx context.Context, tenant filament.TenantID, pipelineID string, includeDeleted bool) ([]*ingestionv1.Notifier, error)

ListNotifiers returns one pipeline's rules ordered by ID.

func (*Store) ListPipelineVersions 

func (s *Store) ListPipelineVersions(ctx context.Context, f filament.PipelineVersionFilter) ([]*ingestionv1.PipelineVersion, int, error)

ListPipelineVersions returns a pipeline's graph versions plus the pre-page total; the default order is newest version first. An unknown pipeline yields an empty slice.

func (*Store) ListPipelines 

func (s *Store) ListPipelines(ctx context.Context, f filament.PipelineFilter) ([]*ingestionv1.Pipeline, int, error)

ListPipelines returns pipelines matching the filter plus the pre-page total.

func (*Store) ListResourceCheckpoints 

func (s *Store) ListResourceCheckpoints(ctx context.Context, tenant filament.TenantID, route filament.ResourceCheckpointRoute) ([]filament.ResourceCheckpointState, error)

ListResourceCheckpoints returns every durable cursor under one route, ordered by resource.

func (*Store) ListResources 

func (s *Store) ListResources(ctx context.Context, tenant filament.TenantID, id filament.RunID) ([]filament.ResourceState, error)

ListResources returns a run's resource states ordered by name.

func (*Store) ListRuns 

func (s *Store) ListRuns(ctx context.Context, f filament.RunFilter) ([]filament.RunState, int, error)

ListRuns returns runs matching the filter, newest effective time first, each with resource states attached, plus the pre-page total. The query is built dynamically, like the postgres store's, because the filter shape doesn't fit a static sqlc statement. Effective time is StartedAt, falling back to RequestedAt, ScheduledAt, then CreatedAt.

func (*Store) ListSchedules 

func (s *Store) ListSchedules(ctx context.Context, f filament.ScheduleFilter) ([]filament.ScheduleState, error)

ListSchedules returns schedules matching the filter.

func (*Store) ListTenants 

func (s *Store) ListTenants(ctx context.Context) ([]Tenant, error)

ListTenants returns all tenant rows ordered by id.

func (*Store) LoadCheckpoint 

func (s *Store) LoadCheckpoint(ctx context.Context, tenant filament.TenantID, id filament.RunID, resource string) (filament.Checkpoint, error)

LoadCheckpoint returns a resource's saved cursor for the run.

func (*Store) LoadConnection 

func (s *Store) LoadConnection(ctx context.Context, tenant filament.TenantID, id string) (filament.Connection, error)

LoadConnection returns the connection with the given ID, including soft-deleted ones so callers can still read a deleted connection's metadata. DeletedAt tells them apart.

func (*Store) LoadNotifier added in v0.3.1 

func (s *Store) LoadNotifier(ctx context.Context, tenant filament.TenantID, pipelineID, id string) (*ingestionv1.Notifier, error)

LoadNotifier returns a rule, including deleted rules for secret cleanup.

func (*Store) LoadPipeline 

func (s *Store) LoadPipeline(ctx context.Context, tenant filament.TenantID, id string) (*ingestionv1.Pipeline, error)

LoadPipeline returns a pipeline by ID, including soft-deleted ones so callers can still read a deleted pipeline's metadata. DeletedAt tells them apart.

func (*Store) LoadPipelineSchedule 

func (s *Store) LoadPipelineSchedule(ctx context.Context, tenant filament.TenantID, pipelineID string) (filament.ScheduleState, error)

LoadPipelineSchedule returns the schedule attached to pipelineID.

func (*Store) LoadPipelineVersion 

func (s *Store) LoadPipelineVersion(ctx context.Context, tenant filament.TenantID, pipelineID string, version int64) (*ingestionv1.PipelineVersion, error)

LoadPipelineVersion returns a specific immutable pipeline graph version; version 0 means the current one.

func (*Store) LoadResourceCheckpoint 

func (s *Store) LoadResourceCheckpoint(ctx context.Context, tenant filament.TenantID, key filament.ResourceCheckpointKey) (filament.ResourceCheckpointState, error)

LoadResourceCheckpoint returns durable cross-run progress for one route and resource.

func (*Store) LoadRun 

func (s *Store) LoadRun(ctx context.Context, tenant filament.TenantID, id filament.RunID) (filament.RunState, error)

LoadRun returns the run with its current resource states reattached.

func (*Store) LoadSchedule 

func (s *Store) LoadSchedule(ctx context.Context, tenant filament.TenantID, id filament.ScheduleID) (filament.ScheduleState, error)

LoadSchedule returns one schedule by id.

func (*Store) LoadTenant 

func (s *Store) LoadTenant(ctx context.Context, id filament.TenantID) (Tenant, error)

LoadTenant returns a tenant row by id.

func (*Store) Name 

func (s *Store) Name() string

Name identifies this store implementation.

func (*Store) Ping 

func (s *Store) Ping(ctx context.Context) error

Ping verifies database reachability.

func (*Store) ReleaseScheduleClaim 

func (s *Store) ReleaseScheduleClaim(ctx context.Context, tenant filament.TenantID, id filament.ScheduleID) error

ReleaseScheduleClaim makes a claimed occurrence eligible for retry.

func (*Store) ResolveTenant 

func (s *Store) ResolveTenant(ctx context.Context, externalID, name string) (filament.TenantID, error)

ResolveTenant maps a provider organization id onto filament's tenant id, minting the row on first sight and refreshing the display name.

func (*Store) SaveCheckpoint 

func (s *Store) SaveCheckpoint(ctx context.Context, tenant filament.TenantID, id filament.RunID, cp filament.Checkpoint) error

SaveCheckpoint upserts a resource's cursor for the run.

func (*Store) SaveResourceCheckpoint 

func (s *Store) SaveResourceCheckpoint(ctx context.Context, tenant filament.TenantID, state filament.ResourceCheckpointState) error

SaveResourceCheckpoint upserts durable cross-run progress for one immutable pipeline route and resource.

func (*Store) SaveRun 

func (s *Store) SaveRun(ctx context.Context, r filament.RunState) error

SaveRun upserts the run row and reattaches any carried Resources into run_resource_states, mirroring the store contract shared with postgres.

func (*Store) SaveSchedule 

func (s *Store) SaveSchedule(ctx context.Context, st filament.ScheduleState) error

SaveSchedule upserts the schedule row, releasing any ClaimDue lease.

func (*Store) TransitionRun 

func (s *Store) TransitionRun(
	ctx context.Context,
	tenant filament.TenantID,
	id filament.RunID,
	from []filament.RunStatus,
	to filament.RunStatus,
	opts filament.RunTransitionOptions,
) (filament.RunState, error)

TransitionRun serializes lifecycle commands on the run row and updates it only when the current status belongs to from. The single-writer connection provides the serialization the postgres store expresses with FOR UPDATE.

func (*Store) UpdateConnection 

func (s *Store) UpdateConnection(ctx context.Context, c filament.Connection) (filament.Connection, error)

UpdateConnection replaces a stored connection, enforcing optimistic version matching.

func (*Store) UpdateNotifier added in v0.3.1 

func (s *Store) UpdateNotifier(ctx context.Context, n *ingestionv1.Notifier) (*ingestionv1.Notifier, error)

UpdateNotifier replaces a live rule's settings.

func (*Store) UpdatePipeline 

func (s *Store) UpdatePipeline(ctx context.Context, p *ingestionv1.Pipeline) (*ingestionv1.Pipeline, error)

UpdatePipeline updates a pipeline's mutable metadata.

func (*Store) UpsertResource 

func (s *Store) UpsertResource(ctx context.Context, rs filament.ResourceState) error

UpsertResource inserts or updates one resource state row.

type Tenant 

type Tenant struct {
	ID        filament.TenantID
	Name      string
	CreatedAt time.Time
	UpdatedAt time.Time
}

Tenant is a control-plane tenant row.

Source Files

View all Source files

Directories

Path Synopsis
Package metrics implements filament.MetricsStore over the same SQLite runs table datastore/sqlite.Store persists to, mirroring datastore/postgres/metrics: a separate package sharing the handle rather than growing the core datastore with dashboard-query-specific SQL.
 Package metrics implements filament.MetricsStore over the same SQLite runs table datastore/sqlite.Store persists to, mirroring datastore/postgres/metrics: a separate package sharing the handle rather than growing the core datastore with dashboard-query-specific SQL.

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.