app

package
v0.14.0 Latest Latest
Warning

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

Go to latest
Published: Oct 29, 2026 License: AGPL-3.0 Imports: 7 Imported by: 0

Documentation

Overview

Package app holds the use cases of Arnakoff and the ports they consume. It depends on domain only (ADR 0002).

Index

Constants

This section is empty.

Variables

View Source
var (
	ErrUnknownSource   = errors.New("unknown source")
	ErrDuplicateSource = errors.New("duplicate source")
)

Sync errors, checked with errors.Is.

View Source
var ErrNoSnapshot = errors.New("no stored snapshot")

ErrNoSnapshot is returned by SnapshotStore.Load when it holds no snapshot of the source.

Functions

This section is empty.

Types

type Clock

type Clock interface {
	Now() time.Time
}

Clock tells the current time; tests inject a fixed one.

type SnapshotStore

type SnapshotStore interface {
	// Save durably replaces the stored snapshot of s.Source.
	Save(ctx context.Context, s domain.Snapshot) error
	// Load returns the stored snapshot of source, or ErrNoSnapshot.
	Load(ctx context.Context, source domain.Source) (domain.Snapshot, error)
}

SnapshotStore keeps the last valid snapshot of each source across restarts and between the sync and API processes.

type SourceFetcher

type SourceFetcher interface {
	// Spec declares the source and what a sane file of it holds.
	Spec() domain.SnapshotSpec
	// Fetch downloads the source and parses it into a snapshot. Sync sets
	// its FetchedAt.
	Fetch(ctx context.Context) (domain.Snapshot, error)
}

SourceFetcher downloads and parses one upstream source.

type SourceSync

type SourceSync struct {
	Source domain.Source
	// LastSuccess is when the snapshot in service was fetched, zero when
	// none is.
	LastSuccess time.Time
	// LastFailure is when the last failed or rejected sync ran, zero when
	// none did. It is kept after a later success.
	LastFailure time.Time
	// LastError is why that sync failed.
	LastError error
	// Rows is the number of rows of the snapshot in service.
	Rows int
	// Newest is the date of the most recent entry in service, zero when the
	// source dates none.
	Newest time.Time
}

SourceSync is the sync state of one source, as /v1/sources shows it.

type Sync

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

Sync downloads each source, validates the new snapshot against the connector's spec and the last valid one, stores it, then swaps it in for the readers. A rejected snapshot leaves the last valid one in service. It is safe for concurrent use.

func NewSync

func NewSync(store SnapshotStore, clock Clock, fetchers ...SourceFetcher) (*Sync, error)

NewSync returns a Sync of the sources the fetchers declare, or ErrDuplicateSource when two declare the same one.

func (*Sync) Restore

func (s *Sync) Restore(ctx context.Context) error

Restore puts in service the stored snapshot of every source when it is valid and newer than the one in service. It runs at startup, and in the API process to pick up what the sync process stored. A source with no stored snapshot is skipped; errors of the others are joined.

func (*Sync) Run

func (s *Sync) Run(ctx context.Context) error

Run syncs every source in turn and joins their errors: one source failing does not stop the others.

func (*Sync) RunSource

func (s *Sync) RunSource(ctx context.Context, source domain.Source) error

RunSource syncs one source, or returns ErrUnknownSource.

func (*Sync) Snapshot

func (s *Sync) Snapshot(source domain.Source) (domain.Snapshot, bool)

Snapshot returns the snapshot in service for source, if any. It is shared with every reader: treat it as read-only.

func (*Sync) States

func (s *Sync) States() []SourceSync

States returns the sync state of every source, in declaration order.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL