store

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Sep 4, 2026 License: Apache-2.0 Imports: 25 Imported by: 0

Documentation

Overview

Package store is Astrate's single PostgreSQL/TimescaleDB access layer (docs/DESIGN.md §1.3): every domain package reads and writes the database through it, and it imports none of them. It owns connection pooling, the embedded schema migrations (docs/DESIGN.md §2.2–2.5), CA-key sealing, and one repository file per aggregate (realms, interfaces, devices, properties, datastreams, groups, triggers) plus LISTEN/NOTIFY plumbing.

Index

Constants

View Source 
const (
	// EnvMasterKey names the env var holding the master key itself,
	// encoded as 64 hex characters or base64 (std or raw) of 32 bytes.
	EnvMasterKey = "ASTRATE_MASTER_KEY"
	// EnvMasterKeyFile names the env var holding a path to a file that
	// contains the master key (raw 32 bytes, or its hex/base64 text form).
	EnvMasterKeyFile = "ASTRATE_MASTER_KEY_FILE"
	// MasterKeySize is the required decoded master key length (AES-256).
	MasterKeySize = 32
)
View Source 
const (
	// DeviceStatusRegistered marks a device that has a credentials secret
	// but has never requested credentials.
	DeviceStatusRegistered = "registered"
	// DeviceStatusConfirmed marks a device that has requested credentials
	// at least once.
	DeviceStatusConfirmed = "confirmed"
	// DeviceStatusInhibited marks a device blocked from new credentials and
	// new connections (credentials_inhibited parity).
	DeviceStatusInhibited = "inhibited"
)

Device status values (devices.status column, docs/DESIGN.md §2.2).

View Source 
const ChannelInterfaces = "astrate_interfaces"

ChannelInterfaces is the NOTIFY channel signalling interface CRUD (docs/DESIGN.md §2.6): the engine LISTENs and rebuilds the affected realm's compiled-interface snapshot. The payload is the realm ID in decimal. In a single process the in-process callback already fires; the channel keeps an optional hot-standby instance coherent.

Variables

View Source 
var (
	// ErrNotFound reports that the addressed row does not exist.
	ErrNotFound = errors.New("store: not found")
	// ErrAlreadyExists reports a uniqueness conflict (realm name, interface
	// name+major, group name, trigger name, ...).
	ErrAlreadyExists = errors.New("store: already exists")
	// ErrInvalidRealmName reports a realm name rejected by the schema's
	// CHECK constraint (must match ^[a-z][a-z0-9]*$).
	ErrInvalidRealmName = errors.New("store: invalid realm name")
	// ErrDeviceAlreadyConfirmed reports a re-registration attempt for a
	// device that has already requested credentials (docs/DESIGN.md §4.4
	// flow A: 422 conflict parity).
	ErrDeviceAlreadyConfirmed = errors.New("store: device has already requested credentials")
	// ErrInterfaceInUse reports an interface delete blocked because some
	// device still declares it in its introspection.
	ErrInterfaceInUse = errors.New("store: interface is referenced by device introspection")
	// ErrInterfaceMajorNotZero reports an interface delete blocked by the
	// upstream draining rule: only major version 0 interfaces are deletable.
	ErrInterfaceMajorNotZero = errors.New("store: only major version 0 interfaces can be deleted")
)

Sentinel errors shared by every repository in this package. Repositories wrap them with context; callers test with errors.Is.

View Source 
var ErrNoMasterKey = fmt.Errorf("store: neither %s nor %s is set", EnvMasterKey, EnvMasterKeyFile)

ErrNoMasterKey reports that neither master-key env reference is set.

View Source 
var ErrPipelineCyclic = errors.New("store: pipeline graph contains a cycle")

ErrPipelineCyclic reports that a pipeline definition's block graph contains a cycle.

Functions

func LoadMasterKey 

func LoadMasterKey() ([]byte, error)

LoadMasterKey resolves the master key from the environment: EnvMasterKey (hex or base64 text) wins over EnvMasterKeyFile (raw 32 bytes, or hex or base64 text, surrounding whitespace ignored).

Types

type DatastreamBatch 

type DatastreamBatch struct {
	Individual []IndividualRow
	Objects    []ObjectRow
}

DatastreamBatch is one persistence flush: the engine's per-shard micro-batches mix individual and object rows (docs/DESIGN.md §1.4).

type Device 

type Device struct {
	ID                       deviceid.ID
	RealmID                  int16
	CredentialsSecretHash    string
	Status                   string
	Introspection            map[string]InterfaceVersion
	OldIntrospection         map[string]InterfaceVersion
	Aliases                  map[string]string
	Attributes               map[string]string
	CertSerial               *string
	CertAKI                  *string
	FirstRegistration        time.Time
	FirstCredentialsRequest  *time.Time
	LastCredentialsRequestIP *netip.Addr
	LastConnection           *time.Time
	LastDisconnection        *time.Time
	LastSeenIP               *netip.Addr
	Connected                bool
	TotalReceivedMsgs        int64
	TotalReceivedBytes       int64
	PayloadFormatHint        string
}

Device is one devices row (docs/DESIGN.md §2.2).

type DownsamplePoint 

type DownsamplePoint struct {
	Bucket time.Time
	Value  float64
}

DownsamplePoint is one downsampled bucket: the bucket start time and the aggregated numeric value.

type Flow added in v0.2.0 

type Flow struct {
	ID           int64
	RealmID      int16
	Name         string
	PipelineName string
	Config       []byte
	AutoRestart  bool
	Status       string
	ErrorMessage *string
	// FailedBlock is the pipeline block whose fatal runtime failure killed
	// the flow (nil unless status=failed with a known culprit).
	FailedBlock *string
	CreatedAt   time.Time
	UpdatedAt   time.Time
	StartedAt   *time.Time
	StoppedAt   *time.Time
}

Flow is one durable named instance of a pipeline for a realm (issues #40 + #41). PipelineName is resolved at start time against the pipelines table; Config is the JSON object used for ${config.*} substitution.

type FlowRehydrate added in v0.2.0 

type FlowRehydrate struct {
	Flow
	RealmName string
}

FlowRehydrate is a durable auto_restart row plus its realm name for boot.

type Group 

type Group struct {
	ID      int64
	RealmID int16
	Name    string
}

Group is one device group (docs/DESIGN.md §2.2). Membership rows carry the composite (realm_id, device_id) foreign key so a device removal cascades out of its groups automatically.

type IndividualRow 

type IndividualRow struct {
	RealmID     int16
	DeviceID    deviceid.ID
	InterfaceID int64
	EndpointID  int64
	Path        string
	TS          time.Time
	ReceptionTS time.Time

	ValueDouble      *float64
	ValueInteger     *int32
	ValueLonginteger *int64
	ValueBoolean     *bool
	ValueString      *string
	ValueBinaryblob  []byte
	ValueDatetime    *time.Time
	ValueArray       []byte
}

IndividualRow is one individual_datastreams row (docs/DESIGN.md §2.4): exactly one Value* field must be set, matching the endpoint's declared type. ValueArray and the object row Value carry pre-encoded JSON.

type InterfaceVersion 

type InterfaceVersion struct {
	Major int `json:"major"`
	Minor int `json:"minor"`
}

InterfaceVersion is one introspection entry value: {"iface.Name": {"major": 1, "minor": 2}}.

type KeySealer 

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

KeySealer seals and opens small secrets (realm CA private keys) with AES-256-GCM. The sealed form is nonce || ciphertext+tag with a fresh random 96-bit nonce per Seal.

func NewKeySealer 

func NewKeySealer(masterKey []byte) (*KeySealer, error)

NewKeySealer builds a KeySealer from a raw 32-byte master key.

func NewKeySealerFromEnv 

func NewKeySealerFromEnv() (*KeySealer, error)

NewKeySealerFromEnv builds a KeySealer from the environment references (EnvMasterKey first, then EnvMasterKeyFile).

func (*KeySealer) Open 

func (ks *KeySealer) Open(sealed []byte) ([]byte, error)

Open decrypts a Seal-produced box, authenticating it in the process.

func (*KeySealer) Seal 

func (ks *KeySealer) Seal(plaintext []byte) ([]byte, error)

Seal encrypts plaintext, returning nonce || ciphertext+tag.

type NewRealm 

type NewRealm struct {
	Name                              string
	JWTPublicKeysPEM                  []string
	CACertificatePEM                  string
	CAPrivateKeySealed                []byte
	DeviceRegistrationLimit           *int32
	DatastreamMaximumStorageRetention *int64
}

NewRealm carries the fields needed to create a realm. CA material is mandatory: a realm without a CA cannot pair devices.

type Notification 

type Notification struct {
	Channel string
	Payload string
}

Notification is one received LISTEN/NOTIFY event.

type ObjectRow 

type ObjectRow struct {
	RealmID     int16
	DeviceID    deviceid.ID
	InterfaceID int64
	Path        string
	TS          time.Time
	ReceptionTS time.Time
	Value       []byte
}

ObjectRow is one object_datastreams row: an object-aggregated publish on a path prefix, with the last-level keys as one JSON document.

type Pipeline added in v0.2.0 

type Pipeline struct {
	ID         int64
	RealmID    int16
	Name       string
	Definition []byte
	CreatedAt  time.Time
	UpdatedAt  time.Time
}

Pipeline is one stored Flow pipeline (astarte_flow parity, issue #24): a named, realm-scoped DAG of blocks. Definition is the raw pipeline JSON (internal/flow.Pipeline shape: "blocks" + "connections"); this package does not import internal/flow (store imports none of its callers, see the package doc), so the graph shape is decoded locally just far enough to validate it.

type Property 

type Property struct {
	RealmID     int16
	DeviceID    deviceid.ID
	InterfaceID int64
	EndpointID  int64
	Path        string
	Value       []byte
	ValueType   interfaceschema.ValueType
	SetAt       time.Time
}

Property is one last-value-wins property row (docs/DESIGN.md §2.3). Value is the jsonb rendering; ValueType is the endpoint's declared type, kept on the row so the API layer re-encodes precisely (longinteger as decimal string, binaryblob as base64, datetime as RFC 3339).

type PropertyRef 

type PropertyRef struct {
	InterfaceID int64
	Path        string
}

PropertyRef addresses one property of a device: the installed interface plus the concrete path.

type Realm 

type Realm struct {
	ID                                int16
	Name                              string
	JWTPublicKeysPEM                  []string
	CACertificatePEM                  string
	CAPrivateKeySealed                []byte
	DeviceRegistrationLimit           *int32
	DatastreamMaximumStorageRetention *int64
	CreatedAt                         time.Time
}

Realm is one tenancy row (docs/DESIGN.md §1.5, §2.2). CAPrivateKeySealed is the AES-256-GCM box produced by KeySealer; the store never sees the plaintext key (sealing happens in the pairing/CA layer).

type RealmPatch added in v0.2.0 

type RealmPatch struct {
	PatchJWTPublicKeyPEM   bool
	SetJWTPublicKeyPEM     string
	PatchRegistrationLimit bool
	SetRegistrationLimit   int32
	ClearRegistrationLimit bool
	PatchRetention         bool
	SetRetention           int64 // seconds; SetRetention=0 with PatchRetention means clear too
	ClearRetention         bool
}

RealmPatch carries optional realm updates; each Patch* flag gates its value field, and Clear* flags write NULL (Clear beats Set).

type SeriesQuery 

type SeriesQuery struct {
	RealmID     int16
	DeviceID    deviceid.ID
	InterfaceID int64
	Path        string
	Since       *time.Time
	SinceAfter  *time.Time
	To          *time.Time
	Limit       int
	Descending  bool
}

SeriesQuery addresses one series (a concrete path of one device interface) and the time window to read. Boundary semantics follow AppEngine parity: Since is inclusive (ts >= Since), SinceAfter is exclusive (ts > SinceAfter), To is inclusive (ts <= To). Limit 0 means no limit; Descending flips the ts ordering (used for "latest N" queries).

type Store 

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

Store is the shared database handle: a bounded pgx pool over an already-migrated schema. All repository methods hang off it; the zero value is not usable, construct with New.

func New 

func New(ctx context.Context, dsn string) (*Store, error)

New connects to dsn, applies any pending embedded migrations, probes optional capabilities, and returns a ready Store. The pool is capped at defaultPoolMaxConns unless the DSN carries an explicit pool_max_conns.

func (*Store) AddDeviceStats 

func (s *Store) AddDeviceStats(ctx context.Context, realmID int16, id deviceid.ID, msgs, bytes int64) error

AddDeviceStats increments the received message/byte counters.

func (*Store) AddGroupDevice 

func (s *Store) AddGroupDevice(ctx context.Context, groupID int64, realmID int16, deviceID deviceid.ID) error

AddGroupDevice adds a device to a group. The INSERT...SELECT guard ensures the group belongs to realmID, and the composite foreign key validates the device; an unknown group or device yields ErrNotFound, an existing membership ErrAlreadyExists.

func (*Store) AliasValuesTaken added in v0.2.0 

func (s *Store) AliasValuesTaken(ctx context.Context, realmID int16, id deviceid.ID, values []string) (bool, error)

AliasValuesTaken reports whether any device other than id in the realm carries one of the given alias values — the upstream find_all_aliases ownership check behind PATCH /devices/{id}'s alias_already_in_use.

func (*Store) AppendDatastreams 

func (s *Store) AppendDatastreams(ctx context.Context, batch DatastreamBatch) error

AppendDatastreams persists a batch through binary COPY, both tables in one transaction (docs/DESIGN.md §5.3: the broker PUBACKs only after this commits). The hypertables have no unique constraints, so duplicate (series, ts) rows from at-least-once redelivery are tolerated by design.

func (*Store) ApplyGlobalRetention 

func (s *Store) ApplyGlobalRetention(ctx context.Context, d time.Duration) error

ApplyGlobalRetention sets a global drop-chunks retention policy on both datastream hypertables (docs/DESIGN.md §2.5: optional and config-driven; the per-endpoint TTL job remains the fine-grained path). A non-positive d clears any existing policy. The configured duration is re-applied idempotently, so changing it in config and restarting takes effect.

func (*Store) Close 

func (s *Store) Close()

Close releases the connection pool. The Store is unusable afterwards.

func (*Store) CountDevices 

func (s *Store) CountDevices(ctx context.Context, realmID int16) (int64, error)

CountDevices returns the number of devices in a realm (used to enforce device_registration_limit in the pairing layer).

func (*Store) CreateFlow added in v0.2.0 

func (s *Store) CreateFlow(ctx context.Context, realmID int16, name, pipelineName string, config []byte, autoRestart bool) (*Flow, error)

CreateFlow inserts a durable flow row. config defaults to {} when empty. autoRestart is stored as given (service layer defaults omitted to true).

func (*Store) CreateGroup 

func (s *Store) CreateGroup(ctx context.Context, realmID int16, name string) (*Group, error)

CreateGroup inserts a group; a duplicate name yields ErrAlreadyExists.

func (*Store) CreatePipeline added in v0.2.0 

func (s *Store) CreatePipeline(ctx context.Context, realmID int16, name string, definition []byte) (*Pipeline, error)

CreatePipeline validates and stores a pipeline; a duplicate name yields ErrAlreadyExists, an invalid graph yields ErrPipelineCyclic or a plain error for an unresolved block reference.

func (*Store) CreateRealm 

func (s *Store) CreateRealm(ctx context.Context, nr NewRealm) (*Realm, error)

CreateRealm inserts the realm row together with its CA material in one transaction (docs/ROADMAP.md §3.1 file 2.8) and returns the stored realm.

func (*Store) CreateTrigger 

func (s *Store) CreateTrigger(ctx context.Context, realmID int16, name string, definition []byte) (*Trigger, error)

CreateTrigger installs a trigger; a duplicate name yields ErrAlreadyExists.

func (*Store) CreateTriggerPolicy added in v0.2.0 

func (s *Store) CreateTriggerPolicy(ctx context.Context, realmID int16, name string, definition []byte) (*TriggerPolicy, error)

CreateTriggerPolicy installs a policy; a duplicate name yields ErrAlreadyExists.

func (*Store) CreateUserBlock added in v0.2.0 

func (s *Store) CreateUserBlock(ctx context.Context, realmID int16, ub *UserBlock) (*UserBlock, error)

CreateUserBlock stores a user block; a duplicate name yields ErrAlreadyExists. An invalid BlockType is rejected by the DB CHECK and surfaces as a plain wrapped error.

func (*Store) DeleteDevice added in v0.2.0 

func (s *Store) DeleteDevice(ctx context.Context, realmID int16, id deviceid.ID) error

DeleteDevice removes the device row and every stored datum in one transaction (synchronous delete — Astrate is single-process, so upstream's async deletion_in_progress flow is unnecessary; docs/COMPATIBILITY.md). Properties and group memberships cascade via FK; the datastream hypertables are swept explicitly, mirroring DeleteRealm. An unknown device yields ErrNotFound.

func (*Store) DeleteFlow added in v0.2.0 

func (s *Store) DeleteFlow(ctx context.Context, realmID int16, name string) error

DeleteFlow removes a durable flow row.

func (*Store) DeleteGroup 

func (s *Store) DeleteGroup(ctx context.Context, realmID int16, name string) error

DeleteGroup removes a group; membership rows cascade.

func (*Store) DeleteInterface 

func (s *Store) DeleteInterface(ctx context.Context, realmID int16, name string, major int) error

DeleteInterface removes an installed interface, enforcing the upstream draining rules: only major version 0 is deletable, and only while no device declares it in its introspection. Properties cascade through their foreign key; datastream rows (no foreign key, docs/DESIGN.md §2.4) are swept in the same transaction.

func (*Store) DeletePipeline added in v0.2.0 

func (s *Store) DeletePipeline(ctx context.Context, realmID int16, name string) error

DeletePipeline removes a pipeline.

func (*Store) DeleteRealm 

func (s *Store) DeleteRealm(ctx context.Context, name string) error

DeleteRealm removes the realm and everything it owns in one transaction (docs/DESIGN.md §2.1: realm deletion is a transactional cascade). Metadata and properties cascade through foreign keys; the datastream hypertables have no foreign keys (§2.4) and are swept explicitly.

func (*Store) DeleteTrigger 

func (s *Store) DeleteTrigger(ctx context.Context, realmID int16, name string) error

DeleteTrigger removes a trigger.

func (*Store) DeleteTriggerPolicy added in v0.2.0 

func (s *Store) DeleteTriggerPolicy(ctx context.Context, realmID int16, name string) error

DeleteTriggerPolicy removes a policy.

func (*Store) DeleteUserBlock added in v0.2.0 

func (s *Store) DeleteUserBlock(ctx context.Context, realmID int16, name string) error

DeleteUserBlock removes a user block; an unknown name yields ErrNotFound.

func (*Store) DeviceStats added in v0.2.0 

func (s *Store) DeviceStats(ctx context.Context, realmID int16) (total, connected int64, err error)

DeviceStats counts a realm's devices and how many are currently connected (the AppEngine /stats/devices endpoint the dashboard polls).

func (*Store) Downsample 

func (s *Store) Downsample(ctx context.Context, q SeriesQuery, bucket time.Duration) ([]DownsamplePoint, error)

Downsample reduces a numeric individual-datastream series to one averaged point per bucket via TimescaleDB time_bucket (AppEngine downsample_to, docs/DESIGN.md §2.5). Non-numeric rows in the window are ignored.

This is the always-available fallback; the toolkit path lives in DownsampleLTTB which callers prefer when Store.HasToolkitLTTB is true.

func (*Store) DownsampleLTTB added in v0.2.0 

func (s *Store) DownsampleLTTB(ctx context.Context, q SeriesQuery, points int) ([]DownsamplePoint, error)

DownsampleLTTB reduces a numeric individual-datastream series to at most points samples using timescaledb_toolkit's lttb() (Largest-Triangle-Three- Buckets), which preserves the visual shape of the curve by selecting real samples rather than averaging them away.

It requires the toolkit: callers must check Store.HasToolkitLTTB first and fall back to Downsample when it is absent (docs/DESIGN.md §2.5). points must be at least 3 — lttb rejects a smaller resolution outright.

func (*Store) EnforceRealmRetentionCeilings added in v0.2.0 

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

EnforceRealmRetentionCeilings applies every realm's datastream_maximum_storage_retention (#72): datastream rows older than the ceiling are deleted regardless of their interface's own retention policy — a set ceiling caps even no_ttl interfaces, which is Astrate's answer to upstream's write-time TTL clamp (Astrate stores no per-row TTL). Realms without a ceiling are skipped. One statement per hypertable per capped realm (no chunk looping at Astrate's scale); a failure on one realm does not abort the others — the first error is returned after all realms ran.

func (*Store) GetDevice 

func (s *Store) GetDevice(ctx context.Context, realmID int16, id deviceid.ID) (*Device, error)

GetDevice fetches one device.

func (*Store) GetDeviceByAlias 

func (s *Store) GetDeviceByAlias(ctx context.Context, realmID int16, alias string) (*Device, error)

GetDeviceByAlias fetches a device by alias value (any alias tag). If several devices share an alias the lowest device ID wins; the API layer is expected to keep aliases unique per realm.

func (*Store) GetFlow added in v0.2.0 

func (s *Store) GetFlow(ctx context.Context, realmID int16, name string) (*Flow, error)

GetFlow fetches one durable flow by realm and name.

func (*Store) GetGroupByName 

func (s *Store) GetGroupByName(ctx context.Context, realmID int16, name string) (*Group, error)

GetGroupByName fetches one group.

func (*Store) GetInterface 

func (s *Store) GetInterface(ctx context.Context, realmID int16, name string, major int) (*StoredInterface, error)

GetInterface loads one installed interface with its endpoint IDs.

func (*Store) GetPipeline added in v0.2.0 

func (s *Store) GetPipeline(ctx context.Context, realmID int16, name string) (*Pipeline, error)

GetPipeline fetches one pipeline by name.

func (*Store) GetProperty 

func (s *Store) GetProperty(ctx context.Context, realmID int16, deviceID deviceid.ID, interfaceID int64, path string) (*Property, error)

GetProperty fetches one property.

func (*Store) GetRealm 

func (s *Store) GetRealm(ctx context.Context, id int16) (*Realm, error)

GetRealm fetches a realm by ID.

func (*Store) GetRealmByName 

func (s *Store) GetRealmByName(ctx context.Context, name string) (*Realm, error)

GetRealmByName fetches a realm by name.

func (*Store) GetTrigger 

func (s *Store) GetTrigger(ctx context.Context, realmID int16, name string) (*Trigger, error)

GetTrigger fetches one trigger by name.

func (*Store) GetTriggerPolicy added in v0.2.0 

func (s *Store) GetTriggerPolicy(ctx context.Context, realmID int16, name string) (*TriggerPolicy, error)

GetTriggerPolicy fetches one policy by name.

func (*Store) GetUserBlock added in v0.2.0 

func (s *Store) GetUserBlock(ctx context.Context, realmID int16, name string) (*UserBlock, error)

GetUserBlock fetches one user block by name.

func (*Store) HasToolkitLTTB 

func (s *Store) HasToolkitLTTB() bool

HasToolkitLTTB reports whether the timescaledb_toolkit extension (and so lttb-based downsampling) was available at startup.

func (*Store) Health 

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

Health verifies database liveness (readiness probe backend).

func (*Store) IndividualSnapshot 

func (s *Store) IndividualSnapshot(ctx context.Context, realmID int16, deviceID deviceid.ID, interfaceID int64) ([]IndividualRow, error)

IndividualSnapshot returns the most recent sample for each distinct path of an individual-datastream interface — the AppEngine interface-root snapshot ("data-snapshot") view upstream renders as a nested tree. Paths that never received a sample are absent; the result is ordered by path.

func (*Store) InstallInterface 

func (s *Store) InstallInterface(ctx context.Context, realmID int16, definition []byte) (*StoredInterface, error)

InstallInterface validates definition and inserts the interface row plus one endpoints row per mapping, all in one transaction. The generated columns (name, versions, type, ownership, aggregation) derive from the stored JSON itself. A (realm, name, major) duplicate yields ErrAlreadyExists.

func (*Store) LatestIndividual added in v0.2.0 

func (s *Store) LatestIndividual(ctx context.Context, realmID int16, deviceID deviceid.ID, interfaceID int64, path string) (*IndividualRow, error)

LatestIndividual reads the newest sample of one individual-datastream series (ORDER BY ts DESC LIMIT 1) — the previous-value lookup feeding the engine's value_change*/path_created trigger evaluation. ErrNotFound when the path never received a sample.

func (*Store) ListAutoRestartFlows added in v0.2.0 

func (s *Store) ListAutoRestartFlows(ctx context.Context) ([]FlowRehydrate, error)

ListAutoRestartFlows returns every flow with auto_restart=true across all realms, for process boot rehydrate.

func (*Store) ListDeviceGroups 

func (s *Store) ListDeviceGroups(ctx context.Context, realmID int16, deviceID deviceid.ID) ([]string, error)

ListDeviceGroups returns the names of every group the device belongs to.

func (*Store) ListDeviceGroupsBatch added in v0.2.0 

func (s *Store) ListDeviceGroupsBatch(ctx context.Context, realmID int16, ids []deviceid.ID) (map[deviceid.ID][]string, error)

ListDeviceGroupsBatch resolves group names for a page of devices in one query (the details=true device listing would otherwise go N+1).

func (*Store) ListDevices 

func (s *Store) ListDevices(ctx context.Context, realmID int16, after *deviceid.ID, limit int) ([]Device, error)

ListDevices returns up to limit devices ordered by ID, starting after the optional keyset cursor. Keyset pagination keeps cursors stable while devices are inserted concurrently.

func (*Store) ListFlows added in v0.2.0 

func (s *Store) ListFlows(ctx context.Context, realmID int16) ([]Flow, error)

ListFlows returns every durable flow for a realm ordered by name.

func (*Store) ListGroupDevices 

func (s *Store) ListGroupDevices(ctx context.Context, groupID int64) ([]deviceid.ID, error)

ListGroupDevices returns the IDs of every device in a group, ordered.

func (*Store) ListGroupDevicesPage added in v0.2.0 

func (s *Store) ListGroupDevicesPage(ctx context.Context, groupID int64, offset, limit int) ([]deviceid.ID, error)

ListGroupDevicesPage returns one page of member IDs ordered by device_id, skipping the first offset rows; at most limit rows are returned.

func (*Store) ListGroups 

func (s *Store) ListGroups(ctx context.Context, realmID int16) ([]Group, error)

ListGroups returns every group of a realm ordered by name.

func (*Store) ListPipelines added in v0.2.0 

func (s *Store) ListPipelines(ctx context.Context, realmID int16) ([]Pipeline, error)

ListPipelines returns every pipeline of a realm ordered by name.

func (*Store) ListProperties 

func (s *Store) ListProperties(ctx context.Context, realmID int16, deviceID deviceid.ID, interfaceID int64) ([]Property, error)

ListProperties returns all properties of one device interface, ordered by path.

func (*Store) ListRealms 

func (s *Store) ListRealms(ctx context.Context) ([]Realm, error)

ListRealms returns every realm ordered by name.

func (*Store) ListServerOwnedProperties 

func (s *Store) ListServerOwnedProperties(ctx context.Context, realmID int16, deviceID deviceid.ID) ([]Property, error)

ListServerOwnedProperties returns every property of the device whose interface is server-owned — the payload of the `/control/consumer/properties` resync message (docs/DESIGN.md §3.4).

func (*Store) ListTriggerPolicies added in v0.2.0 

func (s *Store) ListTriggerPolicies(ctx context.Context, realmID int16) ([]TriggerPolicy, error)

ListTriggerPolicies returns every policy of a realm ordered by name.

func (*Store) ListTriggers 

func (s *Store) ListTriggers(ctx context.Context, realmID int16) ([]Trigger, error)

ListTriggers returns every trigger of a realm ordered by name (the engine's trigger-cache load).

func (*Store) ListUserBlocks added in v0.2.0 

func (s *Store) ListUserBlocks(ctx context.Context, realmID int16) ([]UserBlock, error)

ListUserBlocks returns every user block of a realm ordered by name.

func (*Store) Listen 

func (s *Store) Listen(ctx context.Context, channel string) (<-chan Notification, error)

Listen subscribes to a NOTIFY channel on a dedicated connection (LISTEN pins a session, so the pool is not used) and streams notifications until ctx is cancelled, at which point the returned channel is closed. Lost connections are re-dialled with exponential backoff and the LISTEN is re-issued; notifications emitted while disconnected are lost, which is fine for the cache-invalidation use case (the listener reloads from the tables on resubscribe anyway).

func (*Store) LoadRealmInterfaces 

func (s *Store) LoadRealmInterfaces(ctx context.Context, realmID int16) ([]*StoredInterface, error)

LoadRealmInterfaces loads every interface of a realm with endpoint IDs — the input the engine's schema-compiler cache rebuilds from (docs/ROADMAP.md §3.1 file 2.9 "LoadRealm").

func (*Store) NotifyInterfacesChanged 

func (s *Store) NotifyInterfacesChanged(ctx context.Context, realmID int16) error

NotifyInterfacesChanged emits a ChannelInterfaces notification for realmID.

func (*Store) ObjectSeries 

func (s *Store) ObjectSeries(ctx context.Context, q SeriesQuery) ([]ObjectRow, error)

ObjectSeries reads one object-datastream series (the path is the aggregation prefix).

func (*Store) PatchDeviceAliases 

func (s *Store) PatchDeviceAliases(ctx context.Context, realmID int16, id deviceid.ID, patch map[string]*string) error

PatchDeviceAliases merges patch into the alias map: non-nil values add/replace the tag, nil values remove it (JSON Merge Patch semantics).

func (*Store) PatchDeviceAttributes 

func (s *Store) PatchDeviceAttributes(ctx context.Context, realmID int16, id deviceid.ID, patch map[string]*string) error

PatchDeviceAttributes merges patch into the attributes map with the same semantics as PatchDeviceAliases.

func (*Store) PurgeDeviceOwnedExcept 

func (s *Store) PurgeDeviceOwnedExcept(ctx context.Context, realmID int16, deviceID deviceid.ID, keep []PropertyRef) (int64, error)

PurgeDeviceOwnedExcept deletes every device-owned property of the device that is not in keep — the `/control/producer/properties` resync (docs/DESIGN.md §3.3): the device sends the exhaustive list of properties it still holds and the server drops the complement. Server-owned properties are never touched. It returns the number of rows deleted.

func (*Store) RegisterDevice 

func (s *Store) RegisterDevice(ctx context.Context, realmID int16, id deviceid.ID, secretHash string) error

RegisterDevice inserts a device row, or rotates the credentials secret of an existing one that has not yet requested credentials (docs/DESIGN.md §4.4 flow A). Once the device has requested credentials, re-registration fails with ErrDeviceAlreadyConfirmed.

func (*Store) RemoveGroupDevice 

func (s *Store) RemoveGroupDevice(ctx context.Context, groupID int64, realmID int16, deviceID deviceid.ID) error

RemoveGroupDevice removes a device from a group.

func (*Store) Series 

func (s *Store) Series(ctx context.Context, q SeriesQuery) ([]IndividualRow, error)

Series reads one individual-datastream series.

func (*Store) SeriesSpan added in v0.2.0 

func (s *Store) SeriesSpan(ctx context.Context, q SeriesQuery) (first, last time.Time, ok bool, err error)

SeriesSpan reports the timestamps of the oldest and newest samples of an individual-datastream series under q's filters. ok is false when the filters select no rows, in which case first and last are the zero time.

func (*Store) SetDeviceConnected 

func (s *Store) SetDeviceConnected(ctx context.Context, realmID int16, id deviceid.ID, at time.Time, ip netip.Addr) error

SetDeviceConnected records a broker connection.

func (*Store) SetDeviceCredentials 

func (s *Store) SetDeviceCredentials(ctx context.Context, realmID int16, id deviceid.ID, certSerial, certAKI string, requestIP netip.Addr) error

SetDeviceCredentials stamps a freshly issued client certificate on the device row (serial + authority key identifier, docs/DESIGN.md §4.3 latest-serial enforcement) and records the credentials request.

func (*Store) SetDeviceDisconnected 

func (s *Store) SetDeviceDisconnected(ctx context.Context, realmID int16, id deviceid.ID, at time.Time) error

SetDeviceDisconnected records a broker disconnection.

func (*Store) SetDeviceInhibited 

func (s *Store) SetDeviceInhibited(ctx context.Context, realmID int16, id deviceid.ID, inhibited bool) error

SetDeviceInhibited toggles the inhibit flag (credentials_inhibited PATCH parity): true forces status 'inhibited'; false restores 'confirmed' or 'registered' depending on whether credentials were ever requested.

func (*Store) SetPayloadFormatHint 

func (s *Store) SetPayloadFormatHint(ctx context.Context, realmID int16, id deviceid.ID, hint string) error

SetPayloadFormatHint flips the device's preferred outbound payload format ("bson" or "json", docs/DESIGN.md §3.5.4).

func (*Store) SetRealmCA 

func (s *Store) SetRealmCA(ctx context.Context, name, caCertPEM string, caKeySealed []byte) error

SetRealmCA replaces the realm's CA certificate and sealed private key (re-keying flow, docs/DESIGN.md §4.3).

func (*Store) SetRealmJWTPublicKeys 

func (s *Store) SetRealmJWTPublicKeys(ctx context.Context, name string, keysPEM []string) error

SetRealmJWTPublicKeys replaces the realm's JWT public key set (PEM strings).

func (*Store) Stat 

func (s *Store) Stat() *pgxpool.Stat

Stat returns a snapshot of the connection pool statistics (docs/DESIGN.md §5.2: the DB-pool observability gauges read it).

func (*Store) UnregisterDevice 

func (s *Store) UnregisterDevice(ctx context.Context, realmID int16, id deviceid.ID) error

UnregisterDevice makes a device registrable again without losing its data (DELETE /agent/devices parity, docs/DESIGN.md §4.4): the credentials secret and certificate trail are cleared, the row and all stored data stay.

func (*Store) UnsetProperty 

func (s *Store) UnsetProperty(ctx context.Context, realmID int16, deviceID deviceid.ID, interfaceID int64, path string) (bool, error)

UnsetProperty deletes a property row (empty payload with allow_unset, docs/DESIGN.md §2.3). It reports whether a row existed.

func (*Store) UpdateFlowConfig added in v0.2.0 

func (s *Store) UpdateFlowConfig(ctx context.Context, realmID int16, name string, config []byte) (*Flow, error)

UpdateFlowConfig replaces the config snapshot of a durable flow. The caller is responsible for validating it substitutes cleanly against the flow's pipeline; runtime status columns are left untouched.

func (*Store) UpdateFlowRuntime added in v0.2.0 

func (s *Store) UpdateFlowRuntime(ctx context.Context, realmID int16, name, status string, errMsg *string, failedBlock *string, startedAt, stoppedAt *time.Time) error

UpdateFlowRuntime persists status / error / timestamps after start, stop, or fail. failedBlock is only meaningful with a failed status; passing nil clears it (every successful start/restart does).

func (*Store) UpdateInterface 

func (s *Store) UpdateInterface(ctx context.Context, realmID int16, definition []byte) (*StoredInterface, error)

UpdateInterface applies a minor-version bump: it replaces the stored definition of the existing (realm, name, major) row and inserts endpoints for newly added mappings only — existing endpoint rows are never touched, keeping their IDs stable. The semantic compatibility check (interfaceschema.CheckMinorUpgrade) is the caller's responsibility; this method still refuses a definition that drops an existing endpoint.

func (*Store) UpdateIntrospection 

func (s *Store) UpdateIntrospection(ctx context.Context, realmID int16, id deviceid.ID, intro map[string]InterfaceVersion) (removed map[string]InterfaceVersion, err error)

UpdateIntrospection replaces the device's introspection. Every (name, major) pair present in the current introspection but missing from the new one is merged into old_introspection (docs/ROADMAP.md §3.1 file 2.10) and returned, so the engine can react (cache eviction, property cleanup).

func (*Store) UpdatePipeline added in v0.2.0 

func (s *Store) UpdatePipeline(ctx context.Context, realmID int16, name string, definition []byte) (*Pipeline, error)

UpdatePipeline validates and replaces a pipeline's definition.

func (*Store) UpdateRealm added in v0.2.0 

func (s *Store) UpdateRealm(ctx context.Context, name string, p RealmPatch) error

UpdateRealm applies the patched fields to one realm in a single UPDATE built from only the touched columns. An unknown realm yields ErrNotFound wrapped like the other realm methods.

func (*Store) UpdateUserBlock added in v0.2.0 

func (s *Store) UpdateUserBlock(ctx context.Context, realmID int16, name, blockType string, source, configSchema []byte) (*UserBlock, error)

UpdateUserBlock replaces a user block's type, source and schema; an unknown name yields ErrNotFound.

func (*Store) UpsertProperty 

func (s *Store) UpsertProperty(ctx context.Context, p Property) error

UpsertProperty inserts or replaces a property value (last-value-wins).

type StoredInterface 

type StoredInterface struct {
	ID          int64
	RealmID     int16
	Name        string
	Major       int
	Minor       int
	Type        interfaceschema.InterfaceType
	Ownership   interfaceschema.Ownership
	Aggregation interfaceschema.Aggregation
	Definition  []byte
	// Endpoints maps the declared endpoint pattern (e.g. "/%{sensor_id}/value")
	// to its endpoints.id. IDs are stable for the lifetime of the
	// (realm, name, major) interface, across minor updates and reloads.
	Endpoints map[string]int64
}

StoredInterface is one installed interface: the raw JSON definition (the source of truth, docs/DESIGN.md §2.2) plus the routing-critical generated columns and the stable endpoint-pattern → endpoint-ID map the schema compiler stamps into CompiledMapping rows. It implements interfaceschema.EndpointIDResolver.

func (*StoredInterface) ResolveEndpoint 

func (si *StoredInterface) ResolveEndpoint(endpoint string) (int64, error)

ResolveEndpoint implements interfaceschema.EndpointIDResolver.

func (*StoredInterface) ResolveInterface 

func (si *StoredInterface) ResolveInterface(name string, major int) (int64, error)

ResolveInterface implements interfaceschema.EndpointIDResolver.

type Trigger 

type Trigger struct {
	ID         int64
	RealmID    int16
	Name       string
	Definition []byte
}

Trigger is one installed trigger (docs/DESIGN.md §2.2): the raw Astarte trigger JSON (simple_triggers + action), executed by the engine. Matching upstream Realm Management, triggers have no update operation — reconfiguration is delete + reinstall.

type TriggerPolicy added in v0.2.0 

type TriggerPolicy struct {
	ID         int64
	RealmID    int16
	Name       string
	Definition []byte
}

TriggerPolicy is one stored trigger delivery policy (upstream 1.1 Realm Management surface): the raw policy JSON. The engine compiles it into its realm snapshot and the trigger executor honours it per delivery; the deviations from upstream are recorded in docs/COMPATIBILITY.md.

type UserBlock added in v0.2.0 

type UserBlock struct {
	ID           int64
	RealmID      int16
	Name         string
	BlockType    string // producer, consumer or producer_consumer (DB CHECK)
	Source       []byte
	ConfigSchema []byte // nil when the column is NULL
	CreatedAt    time.Time
	UpdatedAt    time.Time
}

UserBlock is one stored per-realm user-defined composite block (issue #85, astarte_flow parity): a named producer/consumer body inlined at flow start by the flow engine. ConfigSchema is an optional JSON Schema for the block's params and is nil when the column is NULL.

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.