watchdaemon

package
v0.155.1 Latest Latest
Warning

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

 Go to latest Published: Nov 13, 2026 License: Apache-2.0 Imports: 34 Imported by: 0

Documentation

Overview

Package watchdaemon hosts DataTug's independently composable Watch roles.

Index

Constants

View Source 
const (
	ServiceReadContract        = "watch-service-read/1"
	ServiceReadCatalog         = "catalog"
	ServiceReadResolve         = "resolve"
	ServiceReadMaxRows  uint32 = 100
	ServiceReadMaxBytes uint32 = 262144
	ServiceReadLease           = 10 * time.Second
)

Variables

View Source 
var ErrHostedReconcileStep = fmt.Errorf("%w: a hosted Watch bundle cannot run reconcile steps; run reconciliation from the local agent", watch.ErrForbidden)

ErrHostedReconcileStep refuses a hosted Watch bundle whose Check graph holds a reconcile step. A reconcile step names a local profile that reads every selected local environment; a hosted bundle may read only the operator's bound sources through its pinned queries, so reconciliation runs only from the local agent.

Functions

func ValidateEndpoint 

func ValidateEndpoint(endpoint string) error

ValidateEndpoint permits TLS, or explicit loopback HTTP for local operation. Credentials, fragments and query strings never belong in manifest endpoints.

Types

type Authenticator 

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

func NewAuthenticator 

func NewAuthenticator(credentials []Credential) (*Authenticator, error)

func (*Authenticator) Authenticate 

func (a *Authenticator) Authenticate(r *http.Request, kind string) (Credential, bool)

type Client 

type Client struct {
	Endpoint, Token string
	HTTP            *http.Client
}

func (Client) Bundle added in v0.117.0 

func (c Client) Bundle(ctx context.Context, claim watch.Claim) (CloudExecutionBundle, error)

func (Client) Claim 

func (c Client) Claim(ctx context.Context) (*watch.Claim, error)

func (Client) Complete 

func (c Client) Complete(ctx context.Context, result watch.Completion) error

func (Client) Heartbeat 

func (c Client) Heartbeat(ctx context.Context, id string, fence int64) error

func (Client) Register 

func (c Client) Register(ctx context.Context) error

type CloudBundleDependency added in v0.117.0 

type CloudBundleDependency struct {
	Kind     string `json:"kind"`
	ID       string `json:"id"`
	Revision string `json:"revision"`
}

type CloudBundleQuery added in v0.117.0 

type CloudBundleQuery struct {
	Revision       string               `json:"revision"`
	JSON           json.RawMessage      `json:"json"`
	SourcePin      json.RawMessage      `json:"sourcePin,omitempty"`
	SourceIdentity *CloudSourceIdentity `json:"sourceIdentity,omitempty"`
}

type CloudBundleSource added in v0.117.0 

type CloudBundleSource struct {
	ID         string `json:"id"`
	Kind       string `json:"kind"`
	Origin     string `json:"origin,omitempty"`
	Database   string `json:"database,omitempty"`
	Collection string `json:"collection,omitempty"`
	SourceID   string `json:"sourceId,omitempty"`
}

func ServiceConnectionSource added in v0.139.0 

func ServiceConnectionSource(connectionID string) (CloudBundleSource, error)

ServiceConnectionSource makes the existing immutable descriptor shape for a catalog connection. Its hash is a descriptor ID, never access authority.

type CloudExecutionBundle added in v0.117.0 

type CloudExecutionBundle struct {
	Contract       string                      `json:"contract"`
	Scope          watch.Scope                 `json:"scope"`
	CheckID        string                      `json:"checkId"`
	CheckVersion   string                      `json:"checkVersion"`
	SourceRevision string                      `json:"sourceRevision"`
	BundleDigest   string                      `json:"bundleDigest"`
	Catalog        detective.Catalog           `json:"catalog"`
	Queries        map[string]CloudBundleQuery `json:"queries"`
	Sources        []CloudBundleSource         `json:"sources"`
	Dependencies   []CloudBundleDependency     `json:"dependencies"`
	Artifacts      map[string][]byte           `json:"artifacts,omitempty"`
}

CloudExecutionBundle is the versioned wire contract shared with the hosted facade. Core re-verifies the transitive catalog before evaluation.

func (CloudExecutionBundle) ShippingArtifact added in v0.117.0 

func (b CloudExecutionBundle) ShippingArtifact(query CloudBundleQuery) ([]byte, error)

func (CloudExecutionBundle) VerifiedCatalog added in v0.117.0 

func (b CloudExecutionBundle) VerifiedCatalog(ctx context.Context, claim watch.Claim) (detective.Catalog, error)

type CloudRunnerConfig added in v0.117.0 

type CloudRunnerConfig struct {
	SpaceID        string            `json:"spaceID"`
	ProjectID      string            `json:"projectID"`
	SourceBindings map[string]string `json:"sourceBindings"`
}

CloudRunnerConfig is operator-owned. The endpoint remains pinned by the enclosing runner deployment; no cloud token is sent to a project manifest URL. Native bindings name local connections; OVDB bindings name the exact approved origin. Neither bundle nor coordinator receives monitored credentials.

type CloudSourceIdentity added in v0.117.0 

type CloudSourceIdentity struct {
	SourceID        string `json:"sourceId"`
	Environment     string `json:"environment"`
	BindingRevision int64  `json:"bindingRevision"`
	Generation      string `json:"generation"`
	FixtureCommit   string `json:"fixtureCommit"`
	FixtureSHA256   string `json:"fixtureSHA256"`
	SchemaSHA256    string `json:"schemaSHA256"`
}

type CloudSourcePin added in v0.117.0 

type CloudSourcePin struct {
	SpaceID         string `json:"spaceID"`
	ProjectID       string `json:"projectId"`
	SourceID        string `json:"sourceId"`
	Environment     string `json:"environment"`
	BindingRevision int64  `json:"bindingRevision"`
	Revision        struct {
		Generation    string `json:"generation"`
		Revision      int    `json:"revision"`
		SHA256        string `json:"sha256"`
		Bytes         int    `json:"bytes"`
		FixtureCommit string `json:"fixtureCommit"`
		FixtureSHA256 string `json:"fixtureSHA256"`
		SchemaSHA256  string `json:"schemaSHA256"`
	} `json:"revision"`
}

CloudSourcePin contains only fields needed to verify the exact shipping artifact. Other backend revision metadata remains covered by BundleDigest.

type Config 

type Config struct {
	Namespace      string            `json:"namespace"`
	Listen         string            `json:"listen,omitempty"`
	TLSCertFile    string            `json:"tlsCertFile,omitempty"`
	TLSKeyFile     string            `json:"tlsKeyFile,omitempty"`
	LicenseKey     string            `json:"licenseKey,omitempty"`
	Credentials    []Credential      `json:"credentials"`
	AllowedOrigins []string          `json:"allowedOrigins,omitempty"`
	PublicUIURL    string            `json:"publicUIURL,omitempty"`
	ProjectUIURLs  map[string]string `json:"projectUIURLs,omitempty"`
	Limits         Limits            `json:"limits,omitempty"`
	Storage        struct {
		DALGOConnectionEnv string `json:"dalgoConnectionEnv"`
	} `json:"storage"`
	Runner struct {
		Cloud               *CloudRunnerConfig `json:"cloud,omitempty"`
		TokenEnv            string             `json:"tokenEnv"`
		CoordinatorEndpoint string             `json:"coordinatorEndpoint"`
		ProjectDir          string             `json:"projectDir"`
		Environment         string             `json:"environment"`
		Principal           string             `json:"principal"`
		EvidenceDir         string             `json:"evidenceDir"`
		AllowOpaqueSQL      bool               `json:"allowOpaqueSQL,omitempty"`
		Telegram            *TelegramBotConfig `json:"telegram,omitempty"`
	} `json:"runner"`
	Notifications struct {
		EmailSink string      `json:"emailSink,omitempty"`
		SMTP      *SMTPConfig `json:"smtp,omitempty"`
		SMSSink   string      `json:"smsSink,omitempty"`
		Twilio    bool        `json:"twilio,omitempty"`
	} `json:"notifications"`
}

Config belongs to the daemon deployment. Projects carry only coordination endpoint and namespace; credentials and storage are never read from a project.

func LoadConfig 

func LoadConfig(path string) (Config, error)

type CoordinatorReference added in v0.117.0 

type CoordinatorReference struct {
	Endpoint  string `json:"endpoint"`
	Namespace string `json:"namespace"`
}

type CoordinatorStore 

type CoordinatorStore interface {
	Health(context.Context) error
	ListDefinitions(context.Context, watch.Actor) ([]watch.Definition, error)
	GetDefinition(context.Context, watch.Actor, string) (watch.Definition, error)
	PutDefinition(context.Context, watch.Actor, watch.Definition, string, int64) (watch.Definition, error)
	DeleteDefinition(context.Context, watch.Actor, string, string, int64) error
	SetLifecycle(context.Context, watch.Actor, string, string, string, int64) (watch.Definition, error)
	ResetBaseline(context.Context, watch.Actor, string, string, int64) (watch.Definition, error)
	Subscribe(context.Context, watch.Actor, watch.Subscription) error
	GetSubscription(context.Context, watch.Actor, string) (watch.Subscription, error)
	ListRuns(context.Context, watch.Actor, string) ([]watch.Run, error)
	ListDeliveries(context.Context, watch.Actor, string) ([]watch.Delivery, error)
	RetryNotification(context.Context, watch.Actor, string) error
	RegisterRunner(context.Context, watch.RunnerIdentity) error
	ClaimDue(context.Context, watch.RunnerIdentity) (*watch.Claim, error)
	Heartbeat(context.Context, watch.RunnerIdentity, string, int64) error
	FinishRun(context.Context, watch.RunnerIdentity, watch.Completion) (watch.Run, error)
	Inbox(context.Context, watch.Actor) ([]watch.Notification, error)
	MarkRead(context.Context, watch.Actor, string) error
	Status(context.Context, watch.Actor) (watch.Status, error)
	Reconcile(context.Context) error
	ClaimNotifications(context.Context) (*watch.Delivery, error)
	FinishNotification(context.Context, string, int64, string) error
}

CoordinatorStore owns all transactional state; transports never implement scheduling, state transitions, visibility or lease decisions themselves.

type Credential 

type Credential struct {
	Hosted           bool     `json:"hosted,omitempty"`
	SpaceID          string   `json:"spaceID,omitempty"`
	CanManageRunners bool     `json:"canManageRunners,omitempty"`
	TelegramBotID    string   `json:"telegramBotId,omitempty"`
	TokenEnv         string   `json:"tokenEnv"`
	Kind             string   `json:"kind"` // user or runner
	ID               string   `json:"id"`
	AccountID        string   `json:"accountId,omitempty"`
	Namespace        string   `json:"namespace"`
	Project          string   `json:"project"`
	Environment      string   `json:"environment"`
	SharedEditor     bool     `json:"sharedEditor,omitempty"`
	Capabilities     []string `json:"capabilities,omitempty"`
	Email            string   `json:"email,omitempty"`
	Tier             string   `json:"tier,omitempty"`
	Phone            string   `json:"phone,omitempty"`
}

Credential is deployment-owned authority. Tokens are resolved from the environment at startup, never accepted from a project or a request body.

type EmailMessage 

type EmailMessage struct {
	ID      string `json:"id"`
	To      string `json:"to"`
	Subject string `json:"subject"`
	Body    string `json:"body"`
}

EmailMessage contains only notification metadata, never query rows or inputs.

type EmailSender 

type EmailSender interface {
	Send(context.Context, EmailMessage) error
}

type EntitlementChecker added in v0.124.0 

type EntitlementChecker interface {
	CheckEntitlement() error
}

EntitlementChecker validates authorization for centralized/shared operational features.

type Executor 

type Executor interface {
	Execute(context.Context, watch.Claim) (watch.Completion, error)
}

type ExecutorFunc 

type ExecutorFunc func(context.Context, watch.Claim) (watch.Completion, error)

func (ExecutorFunc) Execute 

func (f ExecutorFunc) Execute(ctx context.Context, c watch.Claim) (watch.Completion, error)

type FileEmailSink 

type FileEmailSink struct {
	Path string
	// contains filtered or unexported fields
}

FileEmailSink is a persistent development mailbox. Stable message IDs make retries idempotent, including after a daemon restart. Use one coordinator.

func (*FileEmailSink) Send 

func (s *FileEmailSink) Send(ctx context.Context, message EmailMessage) error

type FileSMSSink 

type FileSMSSink struct{ Sink *FileEmailSink }

FileSMSSink shares the crash-safe, ID-deduplicated development mailbox kernel. Its To field ends in @sms.test and preserves the verified E.164 recipient.

func (FileSMSSink) Send 

func (s FileSMSSink) Send(ctx context.Context, m SMSMessage) error

type Limits 

type Limits struct {
	EmailPerDay        int `json:"emailPerDay,omitempty"`
	EmailPerMinute     int `json:"emailPerMinute,omitempty"`
	SMSPerDay          int `json:"smsPerDay,omitempty"`
	SMSPerMinute       int `json:"smsPerMinute,omitempty"`
	TelegramPerDay     int `json:"telegramPerDay,omitempty"`
	TelegramPerMinute  int `json:"telegramPerMinute,omitempty"`
	WatchesPerUser     int `json:"watchesPerUser,omitempty"`
	MinIntervalSeconds int `json:"minIntervalSeconds,omitempty"`
}

func (Limits) Resolve 

func (l Limits) Resolve() (Limits, error)

type Runner 

type Runner struct {
	Telegram *TelegramBotConfig

	Client                      Client
	Executor                    Executor
	PollInterval, RenewInterval time.Duration
	RegistrationInterval        time.Duration
	// contains filtered or unexported fields
}

func (Runner) Run 

func (r Runner) Run(ctx context.Context) error

Run uses the same HTTP registration/claim/renew/complete protocol locally and remotely. A runner needs no inbound listener or source credentials on server.

type SMSMessage 

type SMSMessage struct{ ID, To, Body string }

type SMSSender 

type SMSSender interface {
	Send(context.Context, SMSMessage) error
}

type SMTPConfig 

type SMTPConfig struct {
	Address     string `json:"address"`
	From        string `json:"from"`
	UsernameEnv string `json:"usernameEnv,omitempty"`
	PasswordEnv string `json:"passwordEnv,omitempty"`
	// AllowLoopbackPlaintext is only for a local development SMTP receiver.
	AllowLoopbackPlaintext bool `json:"allowLoopbackPlaintext,omitempty"`
}

type SMTPSender 

type SMTPSender struct{ Config SMTPConfig }

func (SMTPSender) Send 

func (s SMTPSender) Send(ctx context.Context, m EmailMessage) error

type Server 

type Server struct {
	// ActiveDefinitions applies project discovery visibility after owning Store
	// authorization. It never affects GetDefinition, scheduling or execution.
	ActiveDefinitions func(context.Context, watch.Actor, []watch.Definition) ([]watch.Definition, error)
	// Display/discovery metadata supplied by the trusted hosted facade. It never
	// selects authentication destinations or grants transport authority.
	Coordination *CoordinatorReference
	// ResolveCredential is exclusive when configured. A cloud resolver failure
	// must never fall back to deployment tokens in Auth.
	ResolveCredential func(*http.Request, string) (Credential, error)
	Store             CoordinatorStore
	Auth              *Authenticator
	AllowedOrigins    []string
	Email             EmailSender
	SMS               SMSSender
	PublicUIURL       string
	ProjectUIURLs     map[string]string
	Limits            Limits
	Entitlement       EntitlementChecker
}

func (*Server) CheckEntitlement added in v0.124.0 

func (s *Server) CheckEntitlement() error

CheckEntitlement validates optional entitlement checks for centralized coordinator deployments. In community / standalone mode (s.Entitlement == nil), it returns nil without restriction.

func (*Server) RunCoordinator 

func (s *Server) RunCoordinator(ctx context.Context, interval time.Duration) error

RunCoordinator reconciles due schedules separately from assignment and drains the durable delivery outbox. A failed delivery never changes evaluation truth.

func (*Server) RunCoordinatorTick added in v0.115.0 

func (s *Server) RunCoordinatorTick(ctx context.Context, maxDeliveries int) error

RunCoordinatorTick performs one bounded wakeup for a request-owned cloud instance. Durable jobs, claims and delivery acknowledgements stay in Store; retries or duplicate queue messages do not create another evaluation engine.

func (*Server) ServeHTTP 

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

type ServiceReadBinding added in v0.139.0 

type ServiceReadBinding struct {
	Claim      watch.Claim
	Bundle     CloudExecutionBundle
	QueryID    string
	SourceID   string
	Connection dataaccess.ConnectionRef
}

ServiceReadBinding is retained/admitted metadata supplied independently by each peer. It is never decoded from an operation request. Building a digest authenticates nothing and does not attest that a worker executed a Check.

type ServiceReadIntent added in v0.139.0 

type ServiceReadIntent struct {
	Operation          dataaccess.StructuredRead
	Request            dataaccess.ResultRequest
	WatchBindingSHA256 string
}

func BuildServiceReadIntent added in v0.139.0 

func BuildServiceReadIntent(ctx context.Context, b ServiceReadBinding, parameters map[string]apicontract.TypedValueOrSet, nonce string) (ServiceReadIntent, error)

BuildServiceReadIntent preserves Core's operation-only RequestFor digest and separately binds the complete accepted Watch intent. Query/default codecs are shared by client/server. Bounds are FIXED, not inferred from Detective's500 row limit. Core still owns query admission, policy compilation and output. This helper does not validate Check-step bindings or derive parameters from retained root inputs and nested Check occurrences. Server integration must independently derive and compare parameters for the accepted occurrence using the existing engine's mapping rules, never admit request.Parameters directly.

type ServiceReadRequest added in v0.139.0 

type ServiceReadRequest struct {
	Contract   string                                 `json:"contract"`
	Phase      string                                 `json:"phase"`
	QueryID    string                                 `json:"queryId,omitempty"`
	Parameters map[string]apicontract.TypedValueOrSet `json:"parameters,omitempty"`
	Request    *dataaccess.ResultRequest              `json:"request,omitempty"`
}

ServiceReadRequest is an additive runner/bundle payload, not an identity or permission assertion. Catalog has only contract/phase. Resolve adds operation intent; the server independently admits the runner, claim and accepted query. No endpoint consumes this foundation until its separate integration lands.

func (ServiceReadRequest) MarshalJSON added in v0.139.0 

func (r ServiceReadRequest) MarshalJSON() ([]byte, error)

func (*ServiceReadRequest) UnmarshalJSON added in v0.139.0 

func (r *ServiceReadRequest) UnmarshalJSON(data []byte) error

func (ServiceReadRequest) Validate added in v0.139.0 

func (r ServiceReadRequest) Validate() error

type ServiceReadResponse added in v0.139.0 

type ServiceReadResponse struct {
	Contract           string                 `json:"contract"`
	Phase              string                 `json:"phase"`
	Mode               string                 `json:"mode,omitempty"`
	ProtectedSources   []string               `json:"protectedSources,omitempty"`
	QueryID            string                 `json:"queryId,omitempty"`
	QueryRevision      string                 `json:"queryRevision,omitempty"`
	BundleDigest       string                 `json:"bundleDigest,omitempty"`
	SourceID           string                 `json:"sourceId,omitempty"`
	EnrollmentRevision int64                  `json:"enrollmentRevision,omitempty"`
	RunCeiling         time.Time              `json:"runCeiling,omitempty"`
	Nonce              string                 `json:"nonce,omitempty"`
	WatchBindingSHA256 string                 `json:"watchBindingSHA256,omitempty"`
	Resolution         *dataaccess.Resolution `json:"resolution,omitempty"`
}

ServiceReadResponse contains either catalog classification (no authority) or an existing Core resolution. Decoding checks shape only, not authenticity, current admission, expected Service identity, native identity or continuity. A consumer must independently match retained intent and enter Core's boundary.

func (ServiceReadResponse) MarshalJSON added in v0.139.0 

func (r ServiceReadResponse) MarshalJSON() ([]byte, error)

func (*ServiceReadResponse) UnmarshalJSON added in v0.139.0 

func (r *ServiceReadResponse) UnmarshalJSON(data []byte) error

func (ServiceReadResponse) Validate added in v0.139.0 

func (r ServiceReadResponse) Validate() error

type TelegramBotConfig 

type TelegramBotConfig struct {
	BotID     string `json:"botID"`
	SecretEnv string `json:"secretEnv,omitempty"`
	TokenEnv  string `json:"tokenEnv,omitempty"`
}

TelegramBotConfig is runner-only deployment configuration. Neither token nor secret is sent to the coordinator or saved in a project.

type TwilioSMSSender 

type TwilioSMSSender struct {
	Credentials twiliosms.Credentials
	Transport   http.RoundTripper
}

TwilioSMSSender reuses the ecosystem's shared SMS provider kernel. Credentials remain deployment-owned. Provider errors are deliberately reduced to a code.

func (TwilioSMSSender) Send 

func (s TwilioSMSSender) Send(ctx context.Context, m SMSMessage) error

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.