operations

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: 37 Imported by: 0

Documentation

Overview

Package operations dispatches and executes definition-scoped operations for integrations

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrGalaRequired indicates the gala dependency is missing
	ErrGalaRequired = errors.New("integrations/operations: gala required")
	// ErrDispatchInputInvalid indicates the queued operation request failed caller-input validation
	ErrDispatchInputInvalid = errors.New("integrations/operations: dispatch input invalid")
	// ErrInstallationIDRequired indicates the installation identifier is missing
	ErrInstallationIDRequired = errors.New("integrations/operations: installation id required")
	// ErrIntegrationIDRequired indicates resolution requires an explicit integration ID
	ErrIntegrationIDRequired = errors.New("integrations/operations: integration id required")
	// ErrInstallationDefinitionMismatch indicates the resolved installation does not match the requested definition
	ErrInstallationDefinitionMismatch = errors.New("integrations/operations: installation definition mismatch")
	// ErrOperationConfigInvalid indicates queued operation config failed caller-input validation
	ErrOperationConfigInvalid = errors.New("integrations/operations: operation config invalid")
	// ErrRunIDRequired indicates the run identifier is missing
	ErrRunIDRequired = errors.New("integrations/operations: run id required")
	// ErrIngestDefinitionNotFound indicates the operation definition could not be resolved for ingest
	ErrIngestDefinitionNotFound = errors.New("integrations/operations: ingest definition not found")
	// ErrIngestSchemaNotFound indicates the generated ingest schema contract was not found
	ErrIngestSchemaNotFound = errors.New("integrations/operations: ingest schema not found")
	// ErrIngestSchemaNotDeclared indicates the payload schema was not declared in the operation's ingest contracts
	ErrIngestSchemaNotDeclared = errors.New("integrations/operations: ingest schema not declared in contracts")
	// ErrIngestMappingNotFound indicates the definition does not provide a mapping for the emitted payload variant
	ErrIngestMappingNotFound = errors.New("integrations/operations: ingest mapping not found")
	// ErrIngestFilterFailed indicates the CEL filter evaluation failed
	ErrIngestFilterFailed = errors.New("integrations/operations: ingest filter failed")
	// ErrIngestInstallationFilterConfigInvalid indicates the installation filter configuration could not be decoded
	ErrIngestInstallationFilterConfigInvalid = errors.New("integrations/operations: ingest installation filter config invalid")
	// ErrIngestTransformFailed indicates the CEL map evaluation failed
	ErrIngestTransformFailed = errors.New("integrations/operations: ingest transform failed")
	// ErrIngestMappedDocumentInvalid indicates the mapped payload did not satisfy the generated schema contract
	ErrIngestMappedDocumentInvalid = errors.New("integrations/operations: ingest mapped document invalid")
	// ErrIngestUpsertKeyMissing indicates the mapped payload omitted every generated upsert key
	ErrIngestUpsertKeyMissing = errors.New("integrations/operations: ingest upsert key missing")
	// ErrIngestUpsertConflict indicates the generated upsert keys matched more than one record
	ErrIngestUpsertConflict = errors.New("integrations/operations: ingest upsert conflict")
	// ErrIngestUnsupportedSchema indicates the runtime does not yet support the requested generated ingest schema
	ErrIngestUnsupportedSchema = errors.New("integrations/operations: ingest schema unsupported")
	// ErrIngestPersistFailed indicates the mapped record could not be persisted
	ErrIngestPersistFailed = errors.New("integrations/operations: ingest persistence failed")
	// ErrIngestIntegrationUnresolved indicates the integration record could not be resolved for an ingest operation
	ErrIngestIntegrationUnresolved = errors.New("integrations/operations: ingest integration unresolved")

	// ErrIngestIntegrationRemoved indicates the integration installation was removed while ingest record jobs were still queued
	ErrIngestIntegrationRemoved = errors.New("integrations/operations: ingest integration removed")
	// ErrOperationDisabled indicates the operation is disabled for this installation and the reconcile cycle should stop
	ErrOperationDisabled = errors.New("integrations/operations: operation disabled")
	// ErrExportFailed indicates the Drive file export request failed
	ErrExportFailed = errors.New("integrations/operations: file export failed")
	// ErrResultEncode indicates an operation result could not be serialized
	ErrResultEncode = errors.New("integrations/operations: result encode failed")
	// ErrLinkFailed indicates a link operation failed
	ErrLinkFailed = errors.New("integrations/operations: link operation failed")
)
View Source 
var ReconcileTopic = gala.NamespacedTopicFor(gala.IntegrationReconcile, gala.WithUniqueKey(ReconcileUniqueKey))

ReconcileTopic is the durable reconcile topic: the name derives from the envelope type under the reconcile namespace, and every emission carries the loop uniqueness key

Functions

func CompleteRun 

func CompleteRun(ctx context.Context, db *ent.Client, runID string, startedAt time.Time, result RunResult) error

CompleteRun writes the final run outcome

func CreatePendingRun 

func CreatePendingRun(ctx context.Context, db *ent.Client, installation *ent.Integration, operation string, runType enums.IntegrationRunType, config json.RawMessage) (*ent.IntegrationRun, error)

CreatePendingRun inserts one pending run record for a dispatched operation

func Dispatch 

func Dispatch(ctx context.Context, reg *registry.Registry, db *ent.Client, runtime *gala.Gala, req types.DispatchRequest) (types.DispatchResult, error)

Dispatch validates and enqueues one operation execution request. When DispatchRequest.Runtime is true, no DB integration lookup is performed and the client is resolved from the registry at execution time

func Handle 

func Handle[C DocClient](ref types.ClientRef[C], op types.OperationRef[DocumentExport]) types.OperationHandler

Handle adapts the document export to the generic operation registration boundary

func LastSuccessfulRunAt 

func LastSuccessfulRunAt(ctx context.Context, db *ent.Client, integrationID, operationName string) (*time.Time, error)

LastSuccessfulRunAt returns the finish time of the most recent successful run for the given integration and operation, or nil if no successful run exists yet

func LegacyTopicRenames 

func LegacyTopicRenames() map[gala.TopicName]gala.TopicName

LegacyTopicRenames maps the historical reconcile topic to its designated topic

func MarkRunRunning 

func MarkRunRunning(ctx context.Context, db *ent.Client, runID string) error

MarkRunRunning transitions one run to running

func ReconcileDefinition 

func ReconcileDefinition(reg *registry.Registry, handle func(context.Context, ReconcileEnvelope) (int, error), onExhausted func(context.Context, ReconcileEnvelope, error), schedule gala.Schedule) gala.Definition[ReconcileEnvelope]

ReconcileDefinition builds the Gala listener definition driving every recurring operation cycle: installation-bound reconciliation and runtime-bound scheduled operations

func ReconcileUniqueKey 

func ReconcileUniqueKey(e ReconcileEnvelope) string

ReconcileUniqueKey derives the insert-time uniqueness key for one recurring loop, so any emitter of the topic collapses to at most one live loop per installation (or runtime definition) and operation

func RegisterRuntimeListeners 

func RegisterRuntimeListeners(runtime *gala.Gala, reg *registry.Registry, services types.RuntimeServices, operationHandle func(context.Context, Envelope) error, webhookHandle func(context.Context, WebhookEnvelope) error) error

RegisterRuntimeListeners registers the event, webhook, and definition-provided gala listeners for the integration runtime. Adaptive-scheduled pollers (reconcile, scheduled operations) register themselves at the call site via their own Register*Listener functions

func ResolveIntegration 

func ResolveIntegration(ctx context.Context, db *ent.Client, integrationID, ownerID, definitionID string) (*ent.Integration, error)

ResolveIntegration resolves one integration by explicit ID with optional owner and definition cross-checks

func ResolveOwnerIntegration 

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

ResolveOwnerIntegration finds a connected integration for the given definition and owner. When multiple connected integrations exist, the optional prefer function selects among them. Returns empty string with no error when no integration is found, allowing the caller to fall through to runtime dispatch

func Run 

func Run(ctx context.Context, svc DocClient, cfg *DocumentExport) (json.RawMessage, error)

Run executes the HTML export using the Google Drive API files.export endpoint

func ValidateConfig 

func ValidateConfig(schema json.RawMessage, value json.RawMessage) error

ValidateConfig validates one raw configuration payload against the operation schema

Types

type DocClient 

type DocClient interface {
	Export(ctx context.Context, cfg *DocumentExport) error
}

DocClient is the interface all document integrations must satisfy

type DocumentExport 

type DocumentExport struct {
	// FileID is the external file identifier to export
	FileID string `json:"fileId"`
	// HTML is the exported document content as an embeddable iframe string (populated in the response)
	HTML string `json:"html,omitempty"`
	// PDF is the exported document content as raw PDF bytes (populated in the response)
	PDF []byte `json:"pdf,omitempty"`
	// MimeType is the content type of the downloaded file
	MimeType string `json:"mimeType,omitempty"`
	// Name is the file name without extension
	Name string `json:"name,omitempty"`
}

DocumentExport holds the configuration and result for a integration document export

type Envelope 

type Envelope struct {
	gala.OperationContext
	// Config is the operation configuration payload
	Config json.RawMessage `json:"config,omitempty"`
	// ForceClientRebuild requests client cache bypass
	ForceClientRebuild bool `json:"forceClientRebuild,omitempty"`
}

Envelope is the payload emitted to the operation topic

type IngestContext 

type IngestContext struct {
	// Registry is the integration definition registry used to resolve mappings and definitions
	Registry *registry.Registry
	// DB is the ent client used for persistence
	DB *ent.Client
	// Runtime is the Gala instance used for async emit; nil on the synchronous persist path
	Runtime *gala.Gala
	// Integration is the integration record being ingested into
	Integration *ent.Integration
}

IngestContext holds the stable per-integration dependencies shared across all ingest call paths

type IngestOptions 

type IngestOptions struct {
	// DirectorySyncRunID groups all directory-related ingest records from one sync batch
	DirectorySyncRunID string
	// SkipDirectorySyncRunFinalization instructs the processor not to finalize the directory sync run after processing
	SkipDirectorySyncRunFinalization bool
	// RunID is a caller-supplied correlation identifier for the overall operation run
	RunID string
	// Webhook is the webhook name or identifier that triggered this ingest
	Webhook string
	// WebhookEvent is the event type reported by the webhook provider
	WebhookEvent string
	// DeliveryID is the provider-assigned delivery identifier; used for deduplication
	DeliveryID string
	// WorkflowMeta carries workflow instance context
	WorkflowMeta *types.WorkflowMeta
}

IngestOptions carries the minimal ingest-time metadata needed by persistence

func IngestOptionsFromOperationContext 

func IngestOptionsFromOperationContext(oc gala.OperationContext) IngestOptions

IngestOptionsFromOperationContext derives ingest options from an integration operation context

type IngestResult 

type IngestResult struct {
	Attempted int
	// Persisted counts records written synchronously
	Persisted int
	// Accepted counts records queued durably; acceptance does not mean persistence
	Accepted int
	// Filtered counts records excluded by configured filters
	Filtered int
	// Succeeded is the combined successful handling count
	Succeeded int
	// Failed counts records that could not be imported
	Failed int
	// Failures lists each failed record with its cause
	Failures []RecordFailure
}

IngestResult reports record-level work completed by a payload batch

func EmitPayloadSets 

func EmitPayloadSets(ctx context.Context, ic IngestContext, operationName string, contracts []types.IngestContract, payloadSets []types.IngestPayloadSet, options IngestOptions) (IngestResult, error)

EmitPayloadSets queues each non-directory record for durable schema ingest; directory syncs stay in-process because finalization and removal inference must follow persistence. Queued records report as Accepted, never Persisted

func ProcessPayloadSets 

func ProcessPayloadSets(ctx context.Context, ic IngestContext, operationName string, contracts []types.IngestContract, payloadSets []types.IngestPayloadSet, options IngestOptions) (IngestResult, error)

ProcessPayloadSets persists one batch of mapped payload sets synchronously; record failures are skipped and reported in the result, never the error

type ReconcileEnvelope 

type ReconcileEnvelope struct {
	gala.OperationContext
	// Schedule is the adaptive scheduling state carried across cycles
	Schedule gala.ScheduleState `json:"schedule"`
}

ReconcileEnvelope is the durable payload for one recurring operation cycle, either installation-bound (IntegrationID set) or runtime-bound (Runtime true); the type name is the durable topic identity and must not change

type RecordFailure 

type RecordFailure struct {
	// Schema is the mapping schema name
	Schema string
	// Resource is the provider resource identifier
	Resource string
	// Err is the underlying failure
	Err error
}

RecordFailure identifies one mapped record that could not be imported

type RunResult 

type RunResult struct {
	// Status is the terminal run status
	Status enums.IntegrationRunStatus
	// Summary is the optional summary text stored on the run
	Summary string
	// Error is the optional terminal error text stored on the run
	Error string
	// Metrics is the structured metrics payload stored on the run
	Metrics map[string]any
}

RunResult captures the terminal state of one run

type UserInput 

type UserInput struct {
	// Primary marks this installation as the authoritative source for live document exports
	Primary bool `json:"primary,omitempty" jsonschema:"title=Primary"`
}

UserInput holds installation-specific configuration collected from the user

type WebhookEnvelope 

type WebhookEnvelope struct {
	gala.OperationContext
	// Payload is the raw webhook request body
	Payload json.RawMessage `json:"payload"`
	// Headers contains the inbound HTTP request headers
	Headers map[string]string `json:"headers,omitempty"`
}

WebhookEnvelope is the durable payload emitted for one inbound integration webhook event

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.