jobs

type Access ¶

type Access struct {
	Action Action
	Job    *xasync.Job
	Input  any
}

Access carries a detached public record and, for submit/replay, fresh canonical input. Authorization must revalidate current access and refresh any captured authorization claims on Input. Persisted principal fields alone are not proof.

type Action ¶

type Action string

const (
	Submit  Action = "submit"
	Replay  Action = "replay"
	Inspect Action = "inspect"
)

type Authorizer ¶

type Authorizer func(context.Context, Access) error

type Capture ¶

type Capture struct {
	Plan     *dexec.ReadPlan
	Input    any
	Controls Controls
}

type Config ¶

type Config struct {
	Store     *SQLStore
	Publisher Publisher
	Authorize Authorizer
	TTL       time.Duration
	ErrorTTL  time.Duration
	// Notify receives terminal state only after durable completion write-back.
	// Failure is returned without converting completed work back to ERROR.
	Notify func(context.Context, *xasync.Job) error
}

type Controls ¶

type Controls struct {
	MatchKey string
	JobID    string
	Sync     bool
	Result   bool
}

type Dispatcher ¶

type Dispatcher interface {
	Capture(context.Context, *Record, Submission) (*Capture, error)
	Restore(context.Context, *Record) (Execution, error)
}

Dispatcher adapts jobs to the canonical runtime, never to a second SQL engine.

type Event ¶

type Event struct {
	xasync.Job
	State   string
	Metrics string
	SQL     json.RawMessage
}

Event preserves the original .job document shape, including State and SQL. Public SDK JSON names are case-insensitively readable by original Datly.

type Exchange ¶

type Exchange struct {
	Job    *xasync.Job
	Plan   *dexec.ReadPlan
	Value  any
	Reused bool
}

Exchange is the result of original schedule/sync/inspect orchestration. Value is present only when this invocation actually executed the job.

type Execution ¶

type Execution interface {
	Execute(context.Context) (*ExecutionResult, error)
}

type ExecutionResult ¶

type ExecutionResult struct {
	Value   any
	Outcome *xhandler.Outcome
	Metrics xresponse.Metrics
}

ExecutionResult retains canonical completion evidence without granting transaction handles or completion controls to the job transport.

type InputPolicy ¶

type InputPolicy interface {
	Select(context.Context, any) (Controls, error)
	Owns(*xasync.Job) bool
}

InputPolicy selects scheduling controls from freshly bound canonical input. It is trusted registration metadata, never supplied by a transport client.

type Invocation ¶

type Invocation struct {
	Job  *xasync.Job
	Type xasync.InvocationType
}

Invocation carries the original event-invocation metadata in a fresh replay scope. It is execution metadata, not a global or client-bindable principal.

func CurrentInvocation(ctx context.Context) (Invocation, bool)

func (i Invocation) Context(ctx context.Context) context.Context

type Presentation ¶

type Presentation struct {
	Job   *xasync.Job
	Error error
}

Presentation supplies original Datly KindAsync metadata for one durable job. It holds public detached metadata, never raw SourceState or a principal service.

func (p *Presentation) DefaultCacheable() bool

func (p *Presentation) Kind() string

func (p *Presentation) Locate(*structology.State) locator.Locator

func (p *Presentation) Priority() int

func (p *Presentation) Value(_ context.Context, _ reflect.Type, name string) (any, bool, error)

type PublicationError ¶

type PublicationError struct {
	ID, EventURL string
	Err          error
}

func (e *PublicationError) Error() string

func (e *PublicationError) Unwrap() error

type Publisher ¶

type Publisher interface {
	Prepare(*Record) error
	Publish(context.Context, *Record) error
}

Publisher owns dispatch-event publication, separate from terminal notification.

type Record ¶

type Record struct {
	xasync.Job
	State    string
	Metrics  string
	SQLQuery string
}

Record adds the original persistence-only columns to the existing public SDK model. State is original session-cache JSON keyed by canonical parameter name. SQLQuery uses original Query/Args JSON objects. There is no StateCodec column.

func (r *Record) Event() (*Event, error)

func (r *Record) Public() (*xasync.Job, error)

func (r *Record) QueryScope() (*sqlxread.QueryScope, error)

type ResultReader ¶

type ResultReader interface {
	ReadResult(context.Context, ResultRequest) (any, error)
}

ResultReader is the canonical dispatcher's optional reader capability. It builds a new authorized typed read, never repeats a mutation or decodes output.

type ResultRequest ¶

type ResultRequest struct {
	Job         *Record
	Source      dexec.ProviderScope
	SourceState string
	Refresh     bool
}

type SQLConfig ¶

type SQLConfig struct {
	DB                   *sql.DB
	Dialect              *info.Dialect
	Table                string
	Dataset              string
	DisableTableCreation bool
}

SQLConfig selects the configured job connector and original-shaped job table. Missing tables are created by SQLX unless DisableTableCreation is set. Existing tables are never altered or rewritten.

type SQLStore ¶

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

SQLStore uses the original native SQLX insert/update and typed read owners. Its database is the configured job connector, never the application Data scope.

func NewSQLStore(ctx context.Context, config SQLConfig) (*SQLStore, error)

func (s *SQLStore) Create(ctx context.Context, r *Record) error

func (s *SQLStore) Expire(ctx context.Context, now time.Time) (int64, error)

func (s *SQLStore) Get(ctx context.Context, id string) (*Record, error)

func (s *SQLStore) Pending(ctx context.Context, limit int) ([]*Record, error)

type Scheduled ¶

type Scheduled struct {
	Job  *xasync.Job
	Plan *dexec.ReadPlan
}

type Service ¶

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

func NewService(config Config, dispatcher Dispatcher) (*Service, error)

func (s *Service) Exchange(ctx context.Context, request Submission) (result *Exchange, failure error)

func (s *Service) HandleJob(ctx context.Context, event *Event) (any, error)

func (s *Service) Republish(ctx context.Context, id string) error

func (s *Service) Run(ctx context.Context, id string) (result any, err error)

func (s *Service) Schedule(ctx context.Context, request Submission) (*Scheduled, error)

func (s *Service) Status(ctx context.Context, id string) (*xasync.Job, error)

type StateCodec ¶

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

StateCodec bounds original name-keyed state; all field, presence, source-type and replay binding mechanics belong to the native canonical Bindly plan.

func NewStateCodec(contract *registry.RouteInputContract) (*StateCodec, error)

func (c *StateCodec) Capture(input any) (*bindly.Replay, error)

func (c *StateCodec) CaptureSources(ctx context.Context, providers []locator.Provider) (*bindly.Replay, error)

func (c *StateCodec) Decode(state string) (*bindly.Replay, error)

func (c *StateCodec) Encode(input any) (string, error)

func (c *StateCodec) State(replay *bindly.Replay) (string, error)

type StatusPresentation ¶

type StatusPresentation struct{ Error error }

StatusPresentation supplies the existing output/status shape for metadata-only replies. Actual handler result finalization remains inside the execution engine.

func (p *StatusPresentation) DefaultCacheable() bool

func (p *StatusPresentation) Kind() string

func (p *StatusPresentation) Locate(*structology.State) locator.Locator

func (p *StatusPresentation) Priority() int

func (p *StatusPresentation) Value(_ context.Context, target reflect.Type, name string) (any, bool, error)

type Submission ¶

type Submission struct {
	Job   xasync.Job
	Input any
	// SourceState supplies original parameter-name-keyed raw values for codec
	// inputs or bodies whose authored presence cannot be recovered from a value.
	SourceState string
	Source      dexec.ProviderScope
	Policy      InputPolicy
}