runtime

package
v2.3.0 Latest Latest
Warning

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

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

Documentation

Overview

Package runtime wires the integrations services into one executable runtime

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrInstallationRequired indicates the installation record dependency is missing
	ErrInstallationRequired = errors.New("integrations/runtime: installation required")
	// ErrInstallationNotFound indicates no matching installation could be resolved
	ErrInstallationNotFound = errors.New("integrations/runtime: installation not found")
	// ErrConnectionRequired indicates the installation operation requires a credential-selected connection
	ErrConnectionRequired = errors.New("integrations/runtime: connection required")
	// ErrConnectionNotFound indicates the requested connection could not be resolved for the definition
	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")
	// ErrOperationConfigInvalid indicates the operation config payload failed schema validation
	ErrOperationConfigInvalid = errors.New("integrations/runtime: operation config invalid")
	// 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")
	// ErrCredentialNotDeclared indicates the credential is not declared on the resolved connection
	ErrCredentialNotDeclared = errors.New("integrations/runtime: credential not declared on connection")
	// ErrRuntimeClientNotFound indicates no pre-built runtime client exists for the requested definition
	ErrRuntimeClientNotFound = errors.New("integrations/runtime: runtime client not found")
	// ErrOperationRateLimited indicates the operation's RateLimit policy rejected this run for the organization
	ErrOperationRateLimited = errors.New("integrations/runtime: operation rate limited")
)

Functions

func IntegrationUnhealthyReason 

func IntegrationUnhealthyReason(installation *ent.Integration) string

IntegrationUnhealthyReason returns the recorded reason an installation was marked unhealthy, empty when it is healthy

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 is the server-level development flag; when true, integrations that
	// support it use local file-based senders instead of calling provider APIs
	DevMode bool
	// DefaultLookback sets how far back to fetch data when an operation has no prior successful run;
	// defaults to 90 days when zero
	DefaultLookback time.Duration
}

Config defines the dependencies required to build the integrations runtime

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 PostExecutionHook 

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

PostExecutionHook is called after HandleOperation completes with the processed envelope and any execution 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. Use only in unit tests that exercise registry lookup or operation request wiring.

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, clientID types.ClientID) (any, error)

BuildClientForIntegration builds a typed client for a specific integration installation. It resolves credentials from the keystore and delegates to the registered client builder

func (*Runtime) Catalog 

func (r *Runtime) Catalog() []types.DefinitionSpec

Catalog returns all registered definition specs in stable id order

func (*Runtime) ClearIntegrationUnhealthy 

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

ClearIntegrationUnhealthy returns an errored installation to connected, notifies the owning organization, and reseeds its recurring operations; a non-errored installation is left as is so concurrent recoveries don't stack duplicate notifications

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) Definition 

func (r *Runtime) Definition(id string) (types.Definition, bool)

Definition returns one definition by canonical identifier

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 enqueues one integration operation through the runtime-managed dispatcher

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) (*ent.Integration, bool, error)

EnsureInstallation returns an existing installation when integrationID is provided, or creates a new one

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, credentials types.CredentialBindings, 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 one system-initiated operation inline against a definition's cached runtime client, with no Integration installation and no run tracking. Used for operator-owned calls that need their result back synchronously

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 delta for adaptive scheduling; envelopes with no integration ID run the scheduled runtime path

func (*Runtime) HandleWebhookEvent 

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

HandleWebhookEvent processes one emitted integration webhook envelope

func (*Runtime) LoadCredential 

func (r *Runtime) LoadCredential(ctx context.Context, installation *ent.Integration, credentialRef types.CredentialSlotID) (types.CredentialSet, bool, error)

LoadCredential resolves one persisted credential slot for one installation

func (*Runtime) MarkIntegrationUnhealthy 

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

MarkIntegrationUnhealthy flags one installation as errored with a user-facing reason and notifies the owning organization; recurring cycles stop on their next status check. An installation that is already errored is left as is so repeated failures don't stack duplicate notifications

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 across all job families and returns how many were purged. Operation-context jobs (reconcile loops, event operations) carry the installation as properties.entityId; ingest record jobs carry properties.integration_id

func (*Runtime) Reconcile 

func (r *Runtime) Reconcile(ctx context.Context, installation *ent.Integration, userInput json.RawMessage, credentialRef types.CredentialSlotID, credential *types.CredentialSet, installationInput json.RawMessage) error

Reconcile reconciles installation user input and/or one credential update

func (*Runtime) Redis 

func (r *Runtime) Redis() *redis.Client

Redis returns the shared Redis client from the injector, nil when Redis isn't configured

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 collapses each reconcilable operation on the installation to exactly one recurring loop: an operation already running a single loop is left untouched (preserving its adaptive schedule state), while zero or multiple loops are cancelled and reseeded as one fresh unique loop

func (*Runtime) ResolveIntegration 

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

ResolveIntegration resolves one integration by explicit ID with optional owner and definition cross-checks through the shared operations resolver

func (*Runtime) ResolveOwnerIntegration 

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

ResolveOwnerIntegration finds a connected integration for the given definition and owner through the shared operations resolver

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) SeedReconcileJobs 

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

SeedReconcileJobs ensures every connected integration with reconcilable operations has an active River job. It is intended to be called once at startup to recover reconcile cycles that were lost due to job deletion or a queue flush

func (*Runtime) SeedReconcileJobsForInstallation 

func (r *Runtime) SeedReconcileJobsForInstallation(ctx context.Context, inst *ent.Integration) error

SeedReconcileJobsForInstallation checks every reconcilable operation on the given installation and emits a ReconcileEnvelope for any that do not have an active River job

func (*Runtime) SeedScheduledOperations 

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

SeedScheduledOperations ensures every operation with the Scheduled policy has an active polling loop, called once at startup

func (*Runtime) SetPostExecutionHook 

func (r *Runtime) SetPostExecutionHook(hook PostExecutionHook)

SetPostExecutionHook registers a callback invoked after each HandleOperation call

func (*Runtime) ValidateUserInput 

func (r *Runtime) ValidateUserInput(ctx context.Context, def types.Definition, userInput json.RawMessage) error

ValidateUserInput reports whether a payload satisfies the definition's user input schema, so a caller can reject an install before authorizing rather than failing on every later sync

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.