runtime

package
v2.8.0 Latest Latest
Warning

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

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

Documentation

Overview

Package runtime wires the integrations services into one executable runtime

Index

Constants

View Source 
const (
	// PendingInstallationTTL is how long a pending installation may wait for its auth flow
	PendingInstallationTTL = 168 * time.Hour
)

Variables

View Source 
var (
	// ErrInstallationRequired indicates the installation record dependency is missing
	ErrInstallationRequired = errors.New("integrations/runtime: installation required")
	// ErrIntegrationIDRequired indicates resolution requires an explicit integration ID
	ErrIntegrationIDRequired = errors.New("integrations/runtime: integration id required")
	// ErrInstallationDefinitionMismatch indicates the resolved installation does not match the requested definition
	ErrInstallationDefinitionMismatch = errors.New("integrations/runtime: installation definition mismatch")
	// ErrConnectionRequired indicates the operation requires a credential-selected connection
	ErrConnectionRequired = errors.New("integrations/runtime: connection required")
	// ErrConnectionNotFound indicates the requested connection could not be resolved
	ErrConnectionNotFound = errors.New("integrations/runtime: connection not found")
	// ErrDefinitionNotFound indicates the requested integration definition is not registered
	ErrDefinitionNotFound = errors.New("integrations/runtime: definition not found")
	// ErrOperationNotFound indicates the requested operation is not registered for the definition
	ErrOperationNotFound = errors.New("integrations/runtime: operation not found")
	// ErrUserInputInvalid indicates the user input payload failed schema validation
	ErrUserInputInvalid = errors.New("integrations/runtime: user input invalid")
	// ErrCredentialInvalid indicates the credential payload failed schema validation
	ErrCredentialInvalid = errors.New("integrations/runtime: credential invalid")
	// ErrConnectionMismatch indicates the credential belongs to a different connection than the installation's
	ErrConnectionMismatch = errors.New("integrations/runtime: credential belongs to a different connection than the installation's")
	// ErrInstallationMetadataInvalid indicates the installation metadata document failed schema validation
	ErrInstallationMetadataInvalid = errors.New("integrations/runtime: installation metadata invalid")
	// ErrRuntimeClientNotFound indicates no pre-built runtime client exists for the definition
	ErrRuntimeClientNotFound = errors.New("integrations/runtime: runtime client not found")
	// ErrOperationRateLimited indicates the operation's RateLimit policy rejected this run
	ErrOperationRateLimited = errors.New("integrations/runtime: operation rate limited")
	// ErrInstallationInstanceIDRequired indicates the connection resolved no instance id
	ErrInstallationInstanceIDRequired = errors.New("integrations/runtime: installation instance id required")
	// ErrInstallationInstanceMismatch indicates the credential resolves to a different instance
	ErrInstallationInstanceMismatch = errors.New("integrations/runtime: installation instance mismatch")
	// ErrInstallationVersionAhead indicates a newer binary stamped the installation with a definition version this binary does not have
	ErrInstallationVersionAhead = errors.New("integrations/runtime: installation definition version is ahead of this binary")
	// ErrInstallationUpgradeFailed indicates the installation couldn't reach the current version
	ErrInstallationUpgradeFailed = errors.New("integrations/runtime: installation upgrade failed")
	// ErrClientUnresolved indicates the installation could not establish a connection and needs to be reconnected
	ErrClientUnresolved = errors.New("the integration could not establish a connection and needs to be reconnected")
	// ErrReconcileExhausted indicates a reconcile loop exhausted its error budget
	ErrReconcileExhausted = errors.New("repeated sync failures")
)

Functions

func IngestMetrics 

func IngestMetrics(result operations.IngestResult) map[string]any

IngestMetrics renders one ingest run's record counters as a structured metrics payload

func IngestRunSummary 

func IngestRunSummary(result operations.IngestResult) string

IngestRunSummary renders a compact one-line record-count summary for an ingest run

func RecordFailureSummary 

func RecordFailureSummary(result operations.IngestResult) string

RecordFailureSummary renders a compact summary of a run's failed records

func SetDefault 

func SetDefault(rt *Runtime)

SetDefault registers the process-wide integrations runtime

Types

type Config 

type Config struct {
	// DB is the Ent client used by run stores and direct installation queries
	DB *ent.Client
	// Gala is the event runtime used for operation dispatch and execution
	Gala *gala.Gala
	// Registry overrides the default empty definition registry when provided
	Registry *registry.Registry
	// DefinitionBuilders override the built-in catalog when provided
	DefinitionBuilders []registry.Builder
	// Keystore provides credential persistence and installation-scoped client pooling
	Keystore *keystore.Store
	// RedisClient provides the shared Redis client used for ephemeral integration auth state
	RedisClient *redis.Client
	// CatalogConfig supplies operator-level credentials for all built-in definitions
	CatalogConfig catalog.Config
	// FederationIssuer is the issuer URI customer identity providers federate against
	FederationIssuer string
	// DevMode routes supporting integrations to local file-based senders instead of provider APIs
	DevMode bool
	// DefaultLookback sets the fetch-back window when an operation has no prior successful run
	DefaultLookback time.Duration
}

Config defines the dependencies required to build the integrations runtime

type ConnectionHealthResult added in v2.4.6 

type ConnectionHealthResult struct {
	// Healthy reports whether the connection's credentials passed the health check
	Healthy bool `json:"healthy"`
	// Reason explains the failure when unhealthy
	Reason string `json:"reason,omitempty"`
}

ConnectionHealthResult reports the connection-level health check outcome

type HealthAssessment added in v2.4.6 

type HealthAssessment struct {
	// Status is the installation status after the assessment
	Status enums.IntegrationStatus `json:"status"`
	// Connection is the connection-level check outcome
	Connection ConnectionHealthResult `json:"connection"`
	// Operations lists per-operation probe outcomes and recorded failures
	Operations []OperationHealthResult `json:"operations,omitempty"`
}

HealthAssessment reports the outcome of one installation health run

type IntegrationLookup 

type IntegrationLookup struct {
	// IntegrationID is the unique identifier of the integration installation and required
	IntegrationID string
	// OwnerID scopes the integration to a specific owner, if provided
	OwnerID string
	// DefinitionID validates the integration belongs to a specific definition, if provided
	DefinitionID string
}

IntegrationLookup holds the query constraints for resolving an integration

type OperationHealthResult added in v2.4.6 

type OperationHealthResult struct {
	// Name is the operation name
	Name string `json:"name"`
	// Healthy reports whether the operation's prerequisites are satisfied
	Healthy bool `json:"healthy"`
	// Reason explains the failure when unhealthy
	Reason string `json:"reason,omitempty"`
}

OperationHealthResult reports one operation's health outcome

type PostExecutionHook 

type PostExecutionHook func(ctx context.Context, envelope operations.Envelope, err error)

PostExecutionHook is called after HandleOperation completes with the envelope and any error

type Runtime 

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

Runtime bundles the integrations services behind a do injector

func Default 

func Default() *Runtime

Default returns the process-wide integrations runtime, or nil when none is registered

func New 

func New(config Config) (*Runtime, error)

New wires the integrations runtime

func NewForTesting 

func NewForTesting(reg *registry.Registry) *Runtime

NewForTesting constructs a Runtime backed by the supplied registry and a stub DB client

func (*Runtime) AllowN 

func (r *Runtime) AllowN(ctx context.Context, key string, n int, limit int, window time.Duration) (bool, error)

AllowN reports whether n additional executions fit within limit per window for the namespaced key, consuming n from the window budget when allowed. Keys share the integrations rate limit namespace and always expire after window, so no separate cleanup is required. A server with no redis client configured is never limited

func (*Runtime) BeginAuth 

func (r *Runtime) BeginAuth(ctx context.Context, req keymaker.BeginRequest) (keymaker.BeginResponse, error)

BeginAuth starts one definition auth flow through the runtime-managed keymaker service

func (*Runtime) BuildClientForIntegration 

func (r *Runtime) BuildClientForIntegration(ctx context.Context, integration *ent.Integration, clientName string) (any, error)

BuildClientForIntegration builds a typed client for a specific integration installation

func (*Runtime) ClearIntegrationUnhealthy 

func (r *Runtime) ClearIntegrationUnhealthy(ctx context.Context, installation *ent.Integration) error

ClearIntegrationUnhealthy returns an errored installation to connected and reseeds its loops

func (*Runtime) ClearOperationUnhealthy added in v2.4.6 

func (r *Runtime) ClearOperationUnhealthy(ctx context.Context, installation *ent.Integration, operationName string) error

ClearOperationUnhealthy clears an operation's failure record and resumes its recurring loop

func (*Runtime) CompleteAuth 

func (r *Runtime) CompleteAuth(ctx context.Context, req keymaker.CompleteRequest) (keymaker.CompleteResult, error)

CompleteAuth completes one definition auth flow through the runtime-managed keymaker service

func (*Runtime) DB 

func (r *Runtime) DB() *ent.Client

DB returns the Ent client from the injector

func (*Runtime) Disconnect 

func (r *Runtime) Disconnect(ctx context.Context, installation *ent.Integration) (types.DisconnectResult, error)

Disconnect executes the teardown flow for one installation

func (*Runtime) Dispatch 

func (r *Runtime) Dispatch(ctx context.Context, req types.DispatchRequest) (types.DispatchResult, error)

Dispatch validates and enqueues one operation execution request

func (*Runtime) DispatchWebhookEvent 

func (r *Runtime) DispatchWebhookEvent(ctx context.Context, integration *ent.Integration, definitionID, webhookName string, event types.WebhookReceivedEvent) error

DispatchWebhookEvent emits one normalized integration webhook event through Gala. When integration is nil the event is dispatched as a runtime webhook with no DB-backed installation

func (*Runtime) EnsureInstallation 

func (r *Runtime) EnsureInstallation(ctx context.Context, ownerID, integrationID string, def types.Definition, userInput json.RawMessage, operationConfig map[string]json.RawMessage) (*ent.Integration, bool, error)

EnsureInstallation creates the installation, or resolves the one integrationID names, writing the submitted user input and operation config in one statement; a new installation of a definition without connections is identified by its own id, and one without credentials is created connected

func (*Runtime) EnsureWebhook 

func (r *Runtime) EnsureWebhook(ctx context.Context, integration *ent.Integration, webhookName string, previousIntegrationID string) (*ent.IntegrationWebhook, error)

EnsureWebhook returns the persisted webhook row for one integration and definition webhook

func (*Runtime) ExecuteOperation 

func (r *Runtime) ExecuteOperation(ctx context.Context, integration *ent.Integration, operation types.OperationRegistration, config json.RawMessage) (json.RawMessage, error)

ExecuteOperation runs one integration operation inline without run tracking

func (*Runtime) ExecuteRuntimeOperation 

func (r *Runtime) ExecuteRuntimeOperation(ctx context.Context, definitionID, operationName string, config json.RawMessage) (json.RawMessage, error)

ExecuteRuntimeOperation runs a system-initiated operation inline with no installation

func (*Runtime) FinalizeWebhookDelivery 

func (r *Runtime) FinalizeWebhookDelivery(ctx context.Context, webhook *ent.IntegrationWebhook, deliveryID string, status string, deliveryErr error) error

FinalizeWebhookDelivery updates persisted delivery metadata for one webhook endpoint

func (*Runtime) Gala 

func (r *Runtime) Gala() *gala.Gala

Gala returns the event runtime from the injector

func (*Runtime) HandleOperation 

func (r *Runtime) HandleOperation(ctx context.Context, envelope operations.Envelope) error

HandleOperation executes one queued operation envelope through the runtime-managed dependencies

func (*Runtime) HandleReconcile 

func (r *Runtime) HandleReconcile(ctx context.Context, envelope operations.ReconcileEnvelope) (int, error)

HandleReconcile executes one recurring operation cycle inline and returns the scheduling delta

func (*Runtime) HandleWebhookEvent 

func (r *Runtime) HandleWebhookEvent(ctx context.Context, envelope operations.WebhookEnvelope) error

HandleWebhookEvent processes one emitted integration webhook envelope

func (*Runtime) MarkIntegrationUnhealthy 

func (r *Runtime) MarkIntegrationUnhealthy(ctx context.Context, installation *ent.Integration, reason string) error

MarkIntegrationUnhealthy flags an installation errored and notifies the owning organization

func (*Runtime) MarkOperationUnhealthy added in v2.4.6 

func (r *Runtime) MarkOperationUnhealthy(ctx context.Context, installation *ent.Integration, operationName, reason string) error

MarkOperationUnhealthy records an operation as failing and stops its recurring loop

func (*Runtime) PrepareWebhookDelivery 

func (r *Runtime) PrepareWebhookDelivery(ctx context.Context, webhook *ent.IntegrationWebhook, deliveryID string) (bool, error)

PrepareWebhookDelivery records one delivery idempotency key when present

func (*Runtime) PurgeInstallationJobs 

func (r *Runtime) PurgeInstallationJobs(ctx context.Context, integrationID string) (int, error)

PurgeInstallationJobs removes every queued River job bound to the installation

func (*Runtime) ReapExpiredInstallation added in v2.4.8 

func (r *Runtime) ReapExpiredInstallation(ctx context.Context, integrationID string) (bool, error)

ReapExpiredInstallation soft-deletes an expired never-connected installation and credentials

func (*Runtime) ReconcileCredential 

func (r *Runtime) ReconcileCredential(ctx context.Context, installation *ent.Integration, connectionName string, credential types.CredentialSet) error

ReconcileCredential validates, health-checks, and stores one credential for an installation, re-derives its installation metadata, and activates it

func (*Runtime) Registry 

func (r *Runtime) Registry() *registry.Registry

Registry returns the definition registry

func (*Runtime) ResetReconcileLoops 

func (r *Runtime) ResetReconcileLoops(ctx context.Context, installation *ent.Integration) error

ResetReconcileLoops cancels loops queued under retired operation names and collapses every runnable reconcile operation to exactly one recurring loop

func (*Runtime) ResolveIntegration 

func (r *Runtime) ResolveIntegration(ctx context.Context, lookup IntegrationLookup) (*ent.Integration, error)

ResolveIntegration resolves an integration by ID with optional owner and definition checks

func (*Runtime) ResolveOwnerIntegration 

func (r *Runtime) ResolveOwnerIntegration(ctx context.Context, definitionID, ownerID string, prefer ...func(*ent.Integration) bool) (string, error)

ResolveOwnerIntegration returns the operational installation id for the definition and owner, or empty when none is selectable

func (*Runtime) ResolveWebhookByEndpoint 

func (r *Runtime) ResolveWebhookByEndpoint(ctx context.Context, endpointID string) (*ent.IntegrationWebhook, error)

ResolveWebhookByEndpoint returns the persisted inbound webhook row for the given endpoint ID

func (*Runtime) RunHealthAssessment added in v2.4.6 

func (r *Runtime) RunHealthAssessment(ctx context.Context, installation *ent.Integration) (HealthAssessment, error)

RunHealthAssessment runs the connection and operation health checks and returns the assessment

func (*Runtime) SeedReconcileJobs 

func (r *Runtime) SeedReconcileJobs(ctx context.Context) error

SeedReconcileJobs resets recurring loops for installations with a reconcilable operation

func (*Runtime) SeedScheduledOperations 

func (r *Runtime) SeedScheduledOperations(ctx context.Context) error

SeedScheduledOperations ensures every Scheduled-policy operation has an active polling loop

func (*Runtime) SetPostExecutionHook 

func (r *Runtime) SetPostExecutionHook(hook PostExecutionHook)

SetPostExecutionHook registers a callback invoked after each HandleOperation call

func (*Runtime) UpgradeInstallations 

func (r *Runtime) UpgradeInstallations(ctx context.Context) error

UpgradeInstallations upgrades every connected, degraded, or errored installation whose stored definition version predates the registry version

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.