operations

package
v2.8.2 Latest Latest
Warning

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

 Go to latest Published: Oct 13, 2026 License: Apache-2.0 Imports: 24 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")
	// 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")
	// ErrIngestRecordExcluded indicates the record was skipped because an earlier run recorded it as failing
	ErrIngestRecordExcluded = errors.New("integrations/operations: ingest record excluded")
	// ErrIngestIntegrationUnresolved indicates the integration record could not be resolved for an ingest operation
	ErrIngestIntegrationUnresolved = errors.New("integrations/operations: ingest integration unresolved")
	// ErrIngestInstanceIDRequired indicates the installation carries no source instance id, which every ingested record needs for provenance
	ErrIngestInstanceIDRequired = errors.New("integrations/operations: ingest installation instance id required")

	// 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 types.OperationRegistration, runType enums.IntegrationRunType, config json.RawMessage) (*ent.IntegrationRun, error)

CreatePendingRun inserts one pending run record for a dispatched operation

func ExportDocument 

func ExportDocument[C DocClient](ctx context.Context, _ types.OperationRequest, svc C, cfg DocumentExport) (json.RawMessage, error)

ExportDocument runs the document export against the typed client with the decoded export config

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 MarkRunRunning 

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

MarkRunRunning transitions one run to running

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 RetryRun 

func RetryRun(ctx context.Context, db *ent.Client, run *ent.IntegrationRun) (*ent.IntegrationRun, error)

RetryRun inserts a fresh pending run for a re-executed attempt of the given run, carrying its operation, run type, and config, so every attempt writes records under its own run id

func ValidateInput 

func ValidateInput(ctx context.Context, req types.InstallationRequest, schema json.RawMessage, validate types.ValidateFunc, payload json.RawMessage, sentinel error) error

ValidateInput checks a payload against its schema and then its semantic validation, wrapping schema failures in sentinel

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 {
	// 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 this installation's definition created, changed, or left unchanged
	Persisted int
	// Filtered counts records excluded by configured filters
	Filtered int
	// Succeeded is the combined successful handling count
	Succeeded int
	// Changed counts records that created, modified, or durably queued rows, excluding unchanged rows
	Changed int
	// Skipped counts records left untouched by another definition or a newer run
	Skipped int
	// Failed counts records that could not be imported
	Failed int
	// Excluded counts records that failed again while tracked and were not requeued
	Excluded int
	// Removed counts rows marked removed by snapshot reconciliation
	Removed int
	// Failures lists each failed record with its cause
	Failures []RecordFailure
}

IngestResult reports record-level work completed by a payload batch

func ProcessPayloadSets 

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

ProcessPayloadSets persists one batch of mapped payload sets synchronously

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.