source.go

package filament

import (
	"context"
	"fmt"
	"strings"
	"time"
)

// Source is the connector contract for reading data: describe itself,
// validate and take config, extract into a RecordSink, and tear down.
type Source interface {
	Spec() ConnectorSpec
	Validate(cfg Config) error
	Configure(ctx context.Context, cfg Config) error
	Extract(ctx context.Context, sink RecordSink, opts ExtractOpts) error
	Teardown(ctx context.Context) error
}

// DestinationResourceNamer optionally supplies a default output name without
// schema discovery. Implementations must be deterministic and perform no I/O.
// An empty result leaves the resource name unchanged. Pipeline labels override it.
type DestinationResourceNamer interface {
	DestinationResource(resource string) string
}

// Resumable is the optional source contract for checkpointed extraction. prev maps
// each resource to the cursor to resume from (a plan produced by ResumePlanner,
// carrying any progress from a prior run); a resource absent from prev — or mapped
// to nil — is read from the start. ExtractFrom opens one RowWriter per shard and
// stamps each row's keyset position (RowMeta.Key) so the pipeline can advance and
// persist the cursor.
type Resumable interface {
	ExtractFrom(ctx context.Context, sink RecordSink, opts ExtractOpts, prev map[string]Checkpoint) error
}

// ChangeSource is the optional contract for CDC extraction: emit ordered
// changes from per-resource checkpoints instead of a full read.
type ChangeSource interface {
	ExtractChanges(ctx context.Context, sink RecordSink, opts ChangeExtractOpts) error
}

// ChangeAcknowledger lets a CDC source release upstream stream retention after
// the sink commit and the final per-resource checkpoints are both durable.
// The runner calls it only after observing those checkpoints in the DataStore;
// implementations must never advance beyond the supplied positions.
type ChangeAcknowledger interface {
	AcknowledgeChanges(ctx context.Context, checkpoints map[string]Checkpoint) error
}

// ResourcePlanner lets a source translate externally selected resources into the
// concrete resource names it will emit. Selectors remain source-private; the
// returned names are used by the engine for policy, schema, and checkpoint setup.
type ResourcePlanner interface {
	PlanResources(ctx context.Context, resources, selectors []string) ([]string, error)
}

// ColumnSelector lets a source narrow each resource to the columns a run
// selects, keyed by source resource name. The engine calls it once after
// Configure and before any schema or plan request, so every later read,
// schema, and change stream carries only the selected columns. A resource
// absent from the map reads every column.
type ColumnSelector interface {
	SelectColumns(columns map[string][]string) error
}

// ResumePlanner builds the initial per-resource checkpoint plan for a resumable run:
// the shard layout (PK split) plus any cursor carried over from prev. The engine
// persists the returned plan before extraction so a later resume sees a complete,
// stable shard layout. A resource that cannot be resumed (no primary key) maps to
// nil — the source reads it non-resumably and it is re-read whole on resume.
type ResumePlanner interface {
	PlanResume(ctx context.Context, resources []string, prev map[string]Checkpoint) (map[string]Checkpoint, error)
}

// IncrementalPlanner builds a durable per-resource plan from the previous
// cross-run watermark and versioned cursor configuration.
type IncrementalPlanner interface {
	PlanIncremental(ctx context.Context, resources []string, prev map[string]Checkpoint, cursors map[string]ResourceCursorConfig) (map[string]Checkpoint, error)
}

// IncrementalResumePlanner restores source job state from the current run without
// promoting it to a cross-run watermark. Opt-in plans carry replay_on_resume
// metadata; their rows must be replayed after interruption, even for completed
// jobs, because the previous sink transaction may not have committed.
type IncrementalResumePlanner interface {
	PlanIncrementalResume(ctx context.Context, planned, attempt map[string]Checkpoint) (map[string]Checkpoint, error)
}

// CursorColumnProvider describes which resource columns can safely serve as
// durable incremental cursors and which one the connector recommends.
type CursorColumnProvider interface {
	CursorColumns(ctx context.Context, resource string) ([]CursorColumn, error)
}

// CursorColumn is a schema field annotated with its incremental-cursor
// capabilities. Rank is connector-specific; lower positive values are better.
type CursorColumn struct {
	SchemaField
	PrimaryKey       bool
	Eligible         bool
	Recommended      bool
	Rank             int
	Configurable     bool
	SupportsLookback bool
	Warning          string
}

// ManagedIncrementalSource uses source-owned state rather than a row column.
type ManagedIncrementalSource interface {
	ManagedIncremental(resource string) bool
}

// ResolveCursorVersionPolicy binds a resource's configured or recommended
// incremental cursor into the generic version contract used by sinks. A zero
// policy means the source does not expose cursor metadata and callers should
// retain their insertion-order fallback.
func ResolveCursorVersionPolicy(ctx context.Context, src Source, resource string, config ResourceCursorConfig) (VersionPolicy, error) {
	if managed, ok := src.(ManagedIncrementalSource); ok && managed.ManagedIncremental(resource) {
		if config.Field != "" || config.LookbackSeconds != 0 {
			return VersionPolicy{}, fmt.Errorf("incremental %q uses source-managed state", resource)
		}
		return VersionPolicy{}, nil
	}
	provider, ok := src.(CursorColumnProvider)
	if !ok {
		return VersionPolicy{}, nil
	}
	columns, err := provider.CursorColumns(ctx, resource)
	if err != nil {
		return VersionPolicy{}, fmt.Errorf("cursor columns for %q: %w", resource, err)
	}
	var selected *CursorColumn
	for i := range columns {
		candidate := &columns[i]
		if config.Field != "" {
			if strings.EqualFold(candidate.Name, config.Field) {
				selected = candidate
				break
			}
			continue
		}
		if selected == nil && candidate.Recommended {
			selected = candidate
		}
	}
	if selected == nil {
		if config.Field != "" {
			return VersionPolicy{}, fmt.Errorf("incremental %q cursor column %q does not exist", resource, config.Field)
		}
		return VersionPolicy{}, fmt.Errorf("incremental %q has no recommended cursor column; configure the pipeline resource cursor", resource)
	}
	if !selected.Eligible {
		return VersionPolicy{}, fmt.Errorf("incremental %q cursor column %q is not eligible for durable versioning", resource, selected.Name)
	}
	return VersionPolicy{
		Strategy: VersionCursor,
		Field:    selected.Name,
		Logical:  selected.Logical,
		Native:   selected.Native,
	}, nil
}

// ResolveWriteVersionPolicy binds the version strategy for one source-to-sink
// resource. Non-upsert writes do not need a version; snapshot upserts and
// incremental sources without cursor metadata use insertion order.
func ResolveWriteVersionPolicy(
	ctx context.Context,
	src Source,
	resource string,
	config ResourceCursorConfig,
	sourceMode ReadMode,
	writeMode WriteMode,
) (VersionPolicy, error) {
	if writeMode != WriteUpsert {
		return VersionPolicy{}, nil
	}
	fallback := VersionPolicy{Strategy: VersionInsertOrder}
	if sourceMode != ModeIncremental {
		return fallback, nil
	}
	version, err := ResolveCursorVersionPolicy(ctx, src, resource, config)
	if err != nil {
		return VersionPolicy{}, err
	}
	if version.Strategy == "" {
		return fallback, nil
	}
	return version, nil
}

// Discoverable is the optional contract for browsing a source's available
// resources.
type Discoverable interface {
	Discover(ctx context.Context, opts DiscoverOpts) (DiscoverResult, error)
}

// RateLimited lets a source declare its own extraction rate ceiling.
type RateLimited interface {
	Limits() RatePolicy
}

// ReplicationMode is how a connection replicates, decided at source creation:
// query-based reads or the change stream.
type ReplicationMode string

// The replication modes.
const (
	ReplicationStandard ReplicationMode = "standard"
	ReplicationCDC      ReplicationMode = "cdc"
)

// SnapshotMode is whether a CDC resource without a checkpoint is read in full
// before its change stream starts.
type SnapshotMode string

// The snapshot modes.
const (
	SnapshotInitial SnapshotMode = "initial"
	SnapshotNone    SnapshotMode = "none"
)

// ReplicationStreamPlanningRequest is the connector-owned input for planning
// an independently advancing source consumer.
type ReplicationStreamPlanningRequest struct {
	// Execution selects bounded catch-up or continuous admission constraints.
	Execution ExecutionMode
	// SourceConnectionID is the stable Filament connection identity, independent of runs.
	SourceConnectionID  string
	ReplicationStreamID string
	Config              Config
	// Resources fixes continuous membership. Empty asks the connector for its default.
	Resources []string
}

// ReplicationStreamPlan describes the connector-specific external consumer and
// the normalized config fields that determine whether its continuity is reusable.
type ReplicationStreamPlan struct {
	// Resources is the connector-resolved fixed membership for continuous admission.
	// Bounded CDC planning may leave it empty.
	Resources        []string
	ConsumerName     string
	ConsumerConfig   map[string]any
	ContinuityConfig map[string]any
}

// ReplicationStreamPlanner lets a streaming source own concepts such as a
// PostgreSQL slot, Kafka consumer group, or NATS durable consumer. Binding
// applies an already-resolved stream to the source's runtime configuration.
type ReplicationStreamPlanner interface {
	PlanReplicationStream(request ReplicationStreamPlanningRequest) (ReplicationStreamPlan, error)
	BindReplicationStream(config map[string]any, stream ReplicationStream) (map[string]any, error)
}

// ReplicationStreamCleaner releases a retired connector-side consumer after a
// successor generation has committed successfully. Implementations must treat
// an already-absent consumer as success.
type ReplicationStreamCleaner interface {
	CleanupReplicationStream(ctx context.Context, stream ReplicationStream) error
}

// ReplicationFor resolves a connection's replication mode from the source's
// declared policies and the connection config. CDC requires both: a CDC source
// policy on the spec and replication=cdc in the config.
func ReplicationFor(spec ConnectorSpec, cfg Config) ReplicationMode {
	if cfg.String("replication") != string(ReplicationCDC) {
		return ReplicationStandard
	}
	for _, policy := range spec.SourcePolicies {
		if policy.Mode == ModeCDC {
			return ReplicationCDC
		}
	}
	return ReplicationStandard
}

// LiveValidatable is the optional contract for probing connectivity with a
// config before any run uses it. Both sources and sinks may implement it.
type LiveValidatable interface {
	TestConnection(ctx context.Context, cfg Config) error
}

// ConfigValidatable is the optional contract for connector-specific, pure
// configuration validation. Unlike LiveValidatable it must not access the
// network, so the API can safely use it while accepting connection settings.
type ConfigValidatable interface {
	Validate(cfg Config) error
}

// ConnectorSpec is a source's self-description: identity, supported modes and
// policies, config schema, and resource capabilities. It powers the catalog.
type ConnectorSpec struct {
	// AliasTarget is the concrete registration selected by a catalog alias.
	// The registry sets it; concrete catalog entries leave it empty.
	AliasTarget    string
	Stream         *StreamCapabilities
	Name           string
	DisplayName    string
	Description    string
	DarkLogoURL    string
	LightLogoURL   string
	Version        string
	APIVersion     string
	Maturity       ConnectorMaturity
	Modes          []ReadMode
	SourcePolicies []SourcePolicy
	Config         ConfigSchema
	Resources      ResourceCapabilities
}

// ConfigSchema declares the fields a connector's Config accepts.
type ConfigSchema struct{ Fields []ConfigField }

// ConfigField describes one config field for validation and UI rendering.
type ConfigField struct {
	Name     string
	Type     FieldType
	Required bool
	Default  any
	Enum     []EnumOption
	Help     string
	Scope    FieldScope
	Secret   bool
	// VisibleWhen conditionally includes this field based on a sibling field.
	VisibleWhen *FieldCondition
	// Fields describes the members of an object field. It is empty for scalar
	// fields and for intentionally free-form objects.
	Fields []ConfigField
}

// EnumOption is one ordered choice for an enum or list config field.
type EnumOption struct {
	Value string
	Label string
}

// FieldCondition controls whether a config field applies based on a sibling's
// value. A string sibling matches as is, a bool sibling as "true" or "false".
type FieldCondition struct {
	Field  string
	Values []string
}

// FieldType is a ConfigField's value kind.
type FieldType int

// The config field types.
const (
	FieldString FieldType = iota
	FieldInt
	FieldBool
	FieldSecret // rendered masked; stored as a ref
	FieldDuration
	FieldEnum
	FieldObject
	// FieldList is an ordered list of strings. Enum contains the selectable
	// values when the field is rendered as a checkbox group.
	FieldList
)

// FieldScope is the scope for a config field, aligning with either a connection config or a pipeline config.
// Connection == Only used to establish a connection to connector, reusable for all runs
// Pipeline == Varies based on pipeline
// For example: postgres has a dsn to connect (connection scope) and a table the data should land in (pipeline scope)
type FieldScope int

// The field scopes.
const (
	ScopeUnspecified FieldScope = iota
	ScopeConnection
	ScopePipeline
)

// IsPipeline reports whether the field is set per pipeline node
// rather than on the reusable connection.
func (s FieldScope) IsPipeline() bool { return s == ScopePipeline }

// ReadMode is how a source reads: full scan, incremental from a
// cursor, or CDC.
type ReadMode int

// The replication modes.
const (
	ModeFull ReadMode = iota
	ModeIncremental
	ModeCDC
)

// ResourceCapabilities advertises what a source can do per resource.
type ResourceCapabilities struct{ Discoverable, PerResourceCursor, SelectableColumns bool }

// ExtractOpts scopes one extraction: which resources, how fast. Read behavior
// per resource is carried by the checkpoint plan handed to ExtractFrom, not by
// a run-wide mode.
type ExtractOpts struct {
	Resources   []string
	Selectors   []string
	Limit       int // 0 = unbounded
	Parallelism int
	Observe     SourceObserver
}

// ChangeExtractOpts scopes one CDC extraction: resources plus the checkpoints
// to resume each from.
type ChangeExtractOpts struct {
	Resources   []string
	Checkpoints map[string]Checkpoint
	Limit       int
	Observe     SourceObserver
}

// DiscoverOpts controls discovery; Refresh bypasses any cached catalog.
type DiscoverOpts struct{ Refresh bool }

// DiscoverResult is the resource catalog a Discoverable source returns.
type DiscoverResult struct{ Resources []Resource }

// Resource is one discoverable unit of data (table, stream, endpoint) and the
// metadata needed to select and plan it.
type Resource struct {
	Name        string
	Selector    string
	Selectable  bool
	PrimaryKey  []string
	Schema      *RecordSchema // optional
	Estimated   int64
	DisplayName string
	Metadata    map[string]string
}

// RatePolicy is a token-bucket rate limit on source requests.
type RatePolicy struct {
	RequestsPerSecond float64
	Burst             int
}

// SourceProgressKind identifies a non-terminal extraction signal. Run and
// resource lifecycle state remains runner-owned; these signals describe work
// only the source can observe directly.
type SourceProgressKind uint8

// Source progress kinds identify the source-local signals an observer can report.
const (
	SourceProgressPageFetched SourceProgressKind = iota + 1
	SourceProgressFanOutStarted
	SourceProgressWatermarkAdvanced
	SourceProgressRateLimited
	SourceProgressRetryExhausted
	SourceProgressExportJobCreated
	SourceProgressExportJobPolled
	SourceProgressExportJobReady
)

// ExportJobProgress describes one job execution within an extraction attempt.
// CorrelationID is opaque and changes on resume; it is never an upstream job ID,
// parent key, or URL. Polls and Elapsed are local to this execution, including
// when Resumed is true. Ready means an artifact is available, not sink commit.
type ExportJobProgress struct {
	CorrelationID string        `json:"correlationId"`
	Polls         int           `json:"polls"`
	Elapsed       time.Duration `json:"elapsed"`
	Resumed       bool          `json:"resumed"`
}

// SourceProgress carries source-local extraction progress to the runner. Fields
// are populated according to Kind. Checkpoint on WatermarkAdvanced describes an
// observed cursor only; durable checkpoint persistence remains tracker-owned.
type SourceProgress struct {
	Kind         SourceProgressKind
	Resource     string
	Records      int64
	Bytes        int64
	URI          string
	ParentsTotal int64
	RetryAfter   time.Duration
	Checkpoint   *CheckpointData
	Error        string
	ExportJob    ExportJobProgress
}

// SourceObserver receives progress concurrently when a source extracts more
// than one resource or fans out. Implementations must be concurrency-safe.
type SourceObserver func(SourceProgress)

// Report delivers progress when an observer is configured. A nil observer is
// a safe no-op so sources can report without branching at every call site.
func (o SourceObserver) Report(progress SourceProgress) {
	if o != nil {
		o(progress)
	}
}