Documentation
¶
Overview ¶
Package watch implements project Watches over DALGO transactional state.
Index ¶
- Constants
- Variables
- func ValidateDefinition(definition Definition) error
- type Actor
- type AuditEvent
- type Claim
- type Completion
- type Config
- type Definition
- type DefinitionRepository
- type Delivery
- type Event
- type FirestoreOptions
- type Notification
- type NotificationPolicy
- type Observation
- type RecipientAuthority
- type Run
- type Runner
- type RunnerIdentity
- type Schedule
- type Scope
- type Status
- type Store
- func (s *Store) AdvanceTelegramPoll(ctx context.Context, id RunnerIdentity, fence, offset int64) error
- func (s *Store) Audit(ctx context.Context, a Actor, watchID string) ([]AuditEvent, error)
- func (s *Store) BeginTelegramLink(ctx context.Context, a Actor, nonceHash string, expiresAt time.Time) error
- func (s *Store) ClaimDue(ctx context.Context, identity RunnerIdentity) (*Claim, error)
- func (s *Store) ClaimNotifications(ctx context.Context) (*Delivery, error)
- func (s *Store) ClaimTelegramNotification(ctx context.Context, id RunnerIdentity, pollFence int64) (*Delivery, error)
- func (s *Store) ClaimTelegramPoll(ctx context.Context, id RunnerIdentity) (*TelegramPollLease, error)
- func (s *Store) Close() error
- func (s *Store) CompleteTelegramLinkByRunner(ctx context.Context, id RunnerIdentity, pollFence int64, ...) error
- func (s *Store) DeferTelegramPoll(ctx context.Context, id RunnerIdentity, fence int64, retryAfterSeconds int) error
- func (s *Store) DeleteDefinition(ctx context.Context, a Actor, id, operationID string, expectedVersion int64) error
- func (s *Store) DisconnectTelegram(ctx context.Context, a Actor) error
- func (s *Store) FinishNotification(ctx context.Context, id string, fence int64, failureCode string) error
- func (s *Store) FinishRun(ctx context.Context, identity RunnerIdentity, c Completion) (Run, error)
- func (s *Store) FinishTelegramNotification(ctx context.Context, identity RunnerIdentity, pollFence int64, id string, ...) error
- func (s *Store) GetClaim(ctx context.Context, id RunnerIdentity, runID string, fence int64) (Claim, error)
- func (s *Store) GetDefinition(ctx context.Context, a Actor, id string) (Definition, error)
- func (s *Store) GetEvent(ctx context.Context, a Actor, id string) (Event, error)
- func (s *Store) GetSubscription(ctx context.Context, a Actor, watchID string) (Subscription, error)
- func (s *Store) Health(ctx context.Context) error
- func (s *Store) Heartbeat(ctx context.Context, identity RunnerIdentity, id string, fence int64) error
- func (s *Store) Inbox(ctx context.Context, a Actor) ([]Notification, error)
- func (s *Store) ListDefinitions(ctx context.Context, a Actor) ([]Definition, error)
- func (s *Store) ListDeliveries(ctx context.Context, a Actor, watchID string) ([]Delivery, error)
- func (s *Store) ListEventPage(ctx context.Context, a Actor, watchID, before string, limit int) ([]Event, string, error)
- func (s *Store) ListEvents(ctx context.Context, a Actor, watchID string) ([]Event, error)
- func (s *Store) ListRuns(ctx context.Context, a Actor, watchID string) ([]Run, error)
- func (s *Store) MarkRead(ctx context.Context, a Actor, id string) error
- func (s *Store) NextWakeup(ctx context.Context) (time.Time, bool, error)
- func (s *Store) Prune(ctx context.Context, maxRows int) error
- func (s *Store) PutDefinition(ctx context.Context, a Actor, d Definition, operationID string, ...) (Definition, error)
- func (s *Store) Reconcile(ctx context.Context) error
- func (s *Store) ReconcileScope(ctx context.Context, scope Scope, maxWatches int) (bool, error)
- func (s *Store) RegisterRunner(ctx context.Context, identity RunnerIdentity) error
- func (s *Store) RenewTelegramPoll(ctx context.Context, id RunnerIdentity, fence int64) (*TelegramPollLease, error)
- func (s *Store) ResetBaseline(ctx context.Context, a Actor, id, operationID string, expectedVersion int64) (Definition, error)
- func (s *Store) RetryNotification(ctx context.Context, a Actor, id string) error
- func (s *Store) RevalidateSources(ctx context.Context, scope Scope, maxWatches int, after string) (string, error)
- func (s *Store) SetLifecycle(ctx context.Context, a Actor, id, lifecycle, operationID string, ...) (Definition, error)
- func (s *Store) Status(ctx context.Context, a Actor) (Status, error)
- func (s *Store) Subscribe(ctx context.Context, a Actor, sub Subscription) error
- func (s *Store) TelegramProviderStatus(ctx context.Context, a Actor) (TelegramProviderStatus, error)
- func (s *Store) TelegramStatus(ctx context.Context, a Actor) (bool, error)
- func (s *Store) WithClaim(ctx context.Context, id RunnerIdentity, runID string, fence int64, ...) error
- type Subscription
- type TelegramPollLease
- type TelegramProviderStatus
- type WorkEvent
Constants ¶
const MaxEventPageSize = 1000
Variables ¶
var ( ErrForbidden = errors.New("watch access forbidden") ErrNotFound = errors.New("watch resource not found") ErrConflict = errors.New("watch version conflict") ErrStaleLease = errors.New("watch lease expired or superseded") ErrInvalid = errors.New("invalid watch request") )
var ErrCapacity = errors.New("watch operation exceeds its bounded storage budget")
Functions ¶
func ValidateDefinition ¶ added in v0.119.0
func ValidateDefinition(definition Definition) error
ValidateDefinition validates the native Watch definition without reading or writing a catalog. It lets editors review embedded proposals using the same structural rules as Store.PutDefinition.
This does not authorize saving or execution. PutDefinition must still enforce current scope, source, ownership, subscriptions, capacity and configured limits.
Types ¶
type AuditEvent ¶
type Completion ¶
type Completion struct {
SourceRevision string `json:"sourceRevision,omitempty"`
SourceFingerprint string `json:"sourceFingerprint,omitempty"`
ClaimID string `json:"claimId"`
RunID string `json:"runId"`
Fence int64 `json:"fence"`
ExecutionStatus string `json:"executionStatus"`
Outcome string `json:"outcome"`
Severity string `json:"severity,omitempty"`
DurationMS int64 `json:"durationMs"`
CheckRunRef string `json:"checkRunRef,omitempty"`
FailureCode string `json:"failureCode,omitempty"`
}
type Config ¶
type Config struct {
MaxWatchesPerUser int
MinIntervalSeconds int
TelegramDailyLimit int
TelegramMinuteLimit int
IsPaidAccount func(string) bool
EmailDailyLimit int
SMSDailyLimit int
EmailMinuteLimit int
SMSMinuteLimit int
Namespace string
Now func() time.Time
LeaseDuration time.Duration
RunnerTTL time.Duration
ExecutionTimeout time.Duration
MaxAttempts int
HistoryLimit int
}
type Definition ¶
type Definition struct {
// SourceRevision is a canonical cloud admission pin, not a client grant.
SourceRevision string `json:"sourceRevision,omitempty"`
SourceFingerprint string `json:"sourceFingerprint,omitempty"`
Scope
ID string `json:"id"`
Version int64 `json:"version"`
Title string `json:"title"`
Owner string `json:"owner"`
Visibility string `json:"visibility"`
CheckID string `json:"checkId"`
CheckVersion string `json:"checkVersion,omitempty"`
Inputs map[string]detective.InputValue `json:"inputs,omitempty"`
Schedule Schedule `json:"schedule"`
Trigger string `json:"trigger"` // becomes_true, becomes_false, changes
InitialObservation string `json:"initialObservation"` // notify, baseline
Mode string `json:"mode"` // one_shot, continuous
Lifecycle string `json:"lifecycle"` // enabled, paused, completed, deleted
Enabled bool `json:"enabled"`
Requirements []string `json:"requirements,omitempty"`
NotificationPolicy NotificationPolicy `json:"notificationPolicy"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
type DefinitionRepository ¶
type DefinitionRepository interface {
ListDefinitions(context.Context, Actor) ([]Definition, error)
GetDefinition(context.Context, Actor, string) (Definition, error)
PutDefinition(context.Context, Actor, Definition, string, int64) (Definition, error)
}
DefinitionRepository is durable project storage, independent of job/lease retention. This deployment supplies authenticated DALGO storage; personal definitions are never exported to a project's shared Git files.
type Delivery ¶
type Delivery struct {
TelegramLinkGeneration int64 `json:"telegramLinkGeneration,omitempty"`
TelegramBotID string `json:"telegramBotId,omitempty"`
TelegramChatID string `json:"telegramChatId,omitempty"`
AccountID string `json:"accountId,omitempty"`
Phone string `json:"phone,omitempty"`
QuotaReserved bool `json:"quotaReserved"`
Notification
Channel string `json:"channel"`
Email string `json:"email,omitempty"`
State string `json:"state"`
Attempts int `json:"attempts"`
Fence int64 `json:"fence"`
LeaseUntil time.Time `json:"leaseUntil,omitempty"`
RetryAt time.Time `json:"retryAt,omitempty"`
FailureCode string `json:"failureCode,omitempty"`
}
type Event ¶ added in v0.116.0
type Event struct {
Scope
ID string `json:"id"`
WatchID string `json:"watchId"`
RunID string `json:"runId"`
DefinitionVersion int64 `json:"definitionVersion"`
Title string `json:"title"`
Kind string `json:"kind"` // trigger or recovery
Previous string `json:"previous"` // true, false, never
Outcome string `json:"outcome"` // pass or fail
OccurredAt time.Time `json:"occurredAt"`
CheckID string `json:"checkId"`
CheckVersion string `json:"checkVersion"`
SourceFingerprint string `json:"sourceFingerprint"`
CheckRunRef string `json:"checkRunRef,omitempty"`
Inputs map[string]detective.InputValue `json:"inputs,omitempty"`
Owner string `json:"owner"`
Visibility string `json:"visibility"`
}
Event is the immutable accepted business transition, independent of notification subscriptions and cooldown. Inputs are definition snapshots, never runner rows.
type FirestoreOptions ¶ added in v0.115.0
type FirestoreOptions struct {
ProjectKey *record.Key
// StorageRoot is a separately protected server-only project container. It
// must not be a descendant of a project subtree readable by Space members.
StorageRoot *record.Key
AccountRoot *record.Key
// AccountNamespace is a server-fixed deployment quota namespace. Empty uses Config.Namespace.
AccountNamespace string
ProjectID string
Config Config
Authorize func(context.Context, dal.ReadTransaction) error
AdmitDefinition func(context.Context, dal.ReadTransaction, Definition, string) (string, error)
AdmitRunner func(context.Context, dal.ReadTransaction, Definition, RunnerIdentity) (bool, error)
AdmitRecipient func(context.Context, dal.ReadTransaction, Definition, string) (RecipientAuthority, error)
// DefinitionChanged stages exactly one <=1KiB deterministic metadata event.
// It runs after every read, must not read/publish/network, and shares commit.
DefinitionChanged func(context.Context, dal.ReadwriteTransaction, Definition) error
// WorkReady stages one deterministic metadata wakeup for new pending work.
// It shares DefinitionChanged's no-read/no-network/<=1KiB contract.
WorkReady func(context.Context, dal.ReadwriteTransaction, WorkEvent) error
}
FirestoreOptions binds one Store to an existing trusted project and store/Space namespace. AccountRoot must be shared by projects in that same namespace so creating another project cannot evade account/channel quotas. Both keys and all callbacks are constructed by the authenticated server, never request JSON.
type Notification ¶
type Notification struct {
EventID string `json:"eventId,omitempty"`
Scope
ID string `json:"id"`
WatchID string `json:"watchId"`
RunID string `json:"runId"`
UserID string `json:"userId"`
Title string `json:"title"`
Event string `json:"event"`
Link string `json:"link"`
CreatedAt time.Time `json:"createdAt"`
ReadAt time.Time `json:"readAt,omitempty"`
}
type NotificationPolicy ¶
type Observation ¶
type Observation struct {
LastEventID string `json:"lastEventId,omitempty"`
SourceFingerprint string `json:"sourceFingerprint,omitempty"`
WatchID string `json:"watchId"`
Value string `json:"value"` // true, false, unknown, never
LastKnown string `json:"lastKnown"`
LastRunID string `json:"lastRunId,omitempty"`
LastObservedAt time.Time `json:"lastObservedAt,omitempty"`
LastNotifiedAt time.Time `json:"lastNotifiedAt,omitempty"`
NextEvaluation time.Time `json:"nextEvaluation"`
Availability string `json:"availability"`
}
type RecipientAuthority ¶ added in v0.115.0
RecipientAuthority is fresh server-owned identity and entitlement evidence. Paid is never populated from a browser request or an unverified token claim.
type Run ¶
type Run struct {
EventID string `json:"eventId,omitempty"`
Scope
ID string `json:"id"`
WatchID string `json:"watchId"`
DefinitionVersion int64 `json:"definitionVersion"`
DueAt time.Time `json:"dueAt"`
State string `json:"state"` // due, running, completed, retry, failed, cancelled
RunnerID string `json:"runnerId,omitempty"`
Fence int64 `json:"fence"`
Attempts int `json:"attempts"`
Deadline time.Time `json:"deadline,omitempty"`
CompletionHash string `json:"completionHash,omitempty"`
LeaseUntil time.Time `json:"leaseUntil,omitempty"`
RetryAt time.Time `json:"retryAt,omitempty"`
ExecutionStatus string `json:"executionStatus,omitempty"`
Outcome string `json:"outcome,omitempty"`
Severity string `json:"severity,omitempty"`
DurationMS int64 `json:"durationMs,omitempty"`
CheckRunRef string `json:"checkRunRef,omitempty"`
FailureCode string `json:"failureCode,omitempty"`
FinishedAt time.Time `json:"finishedAt,omitempty"`
}
type RunnerIdentity ¶
type RunnerIdentity struct {
TelegramBotID string
TelegramBotUsername string
ID string
Scope Scope
Capabilities []string
}
RunnerIdentity contains only grants verified by the coordinator transport.
type Status ¶
type Status struct {
Coordinator string `json:"coordinator"`
Runners []Runner `json:"runners"`
Observations []Observation `json:"observations"`
}
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
func NewFirestoreStore ¶ added in v0.115.0
func NewFirestoreStore(db dal.DB, options FirestoreOptions) (*Store, error)
NewFirestoreStore reuses the pinned DALGO Firestore adapter. It neither creates a database nor owns its shared client; Store.Close leaves that client open. Authorization executes inside the same transaction as protected reads/writes. Callback errors abort the transaction, and callbacks may run more than once.
func NewSQLiteStore ¶
NewSQLiteStore accepts an absolute local file path or sqlite:///absolute/path. Storage configuration belongs to daemon deployment. Arbitrary driver DSNs are refused so a caller cannot override transaction locking or durability pragmas.
func (*Store) AdvanceTelegramPoll ¶
func (*Store) BeginTelegramLink ¶
func (s *Store) BeginTelegramLink(ctx context.Context, a Actor, nonceHash string, expiresAt time.Time) error
BeginTelegramLink binds a short-lived connection request to the authenticated user, project scope and one registered bot. No browser supplies a chat ID.
func (*Store) ClaimNotifications ¶
ClaimNotifications is an internal coordinator worker API. HTTP must never expose it to browser users or runners: it contains delivery-only addresses.
func (*Store) ClaimTelegramNotification ¶
func (*Store) ClaimTelegramPoll ¶
func (s *Store) ClaimTelegramPoll(ctx context.Context, id RunnerIdentity) (*TelegramPollLease, error)
func (*Store) CompleteTelegramLinkByRunner ¶
func (s *Store) CompleteTelegramLinkByRunner(ctx context.Context, id RunnerIdentity, pollFence int64, nonceHash, chatID string) error
CompleteTelegramLinkByRunner accepts proof only from the fenced provider poller whose operator grant owns this bot and scope. Replayed messages for the same chat are idempotent; another chat never steals a consumed connection.
func (*Store) DeferTelegramPoll ¶
func (s *Store) DeferTelegramPoll(ctx context.Context, id RunnerIdentity, fence int64, retryAfterSeconds int) error
DeferTelegramPoll persists Telegram's bot-wide flood wait and fences all old requests, so failover to a second runner cannot bypass the provider cooldown.
func (*Store) DeleteDefinition ¶
func (*Store) DisconnectTelegram ¶
func (*Store) FinishNotification ¶
func (*Store) FinishRun ¶
func (s *Store) FinishRun(ctx context.Context, identity RunnerIdentity, c Completion) (Run, error)
func (*Store) FinishTelegramNotification ¶
func (*Store) GetClaim ¶ added in v0.115.0
func (s *Store) GetClaim(ctx context.Context, id RunnerIdentity, runID string, fence int64) (Claim, error)
GetClaim validates the current lease, definition version, runner eligibility, owner/source admission and baseline in one transaction before a bundle read.
func (*Store) GetDefinition ¶
func (*Store) GetEvent ¶ added in v0.116.0
GetEvent checks current visibility in the same transaction as reading evidence. Historical personal inputs never gain access through a stale authorization read.
func (*Store) GetSubscription ¶
func (*Store) Health ¶
Health reads persisted storage, so an unavailable file is not reported healthy.
func (*Store) ListDefinitions ¶
func (*Store) ListDeliveries ¶
func (*Store) ListEventPage ¶ added in v0.116.0
func (s *Store) ListEventPage(ctx context.Context, a Actor, watchID, before string, limit int) ([]Event, string, error)
ListEventPage bounds database work and returns newest first. The opaque cursor is an event ID in this exact Watch and scope; timestamp ties use storage IDs. Events are durable evidence and deliberately excluded from runtime pruning.
func (*Store) ListEvents ¶ added in v0.116.0
func (*Store) NextWakeup ¶ added in v0.115.0
NextWakeup returns the next bounded scheduling/delivery wakeup. No work means a scale-to-zero queue handler can stop scheduling tasks. Telegram polling is runner-owned and does not keep a coordinator task chain alive.
func (*Store) Prune ¶ added in v0.115.0
Prune is bounded maintenance for terminal history and expired helper records. Pending jobs/deliveries are never selected, so an accumulated history cannot wedge claim/complete operations or silently discard unpublished work.
func (*Store) PutDefinition ¶
func (s *Store) PutDefinition(ctx context.Context, a Actor, d Definition, operationID string, expectedVersion int64) (Definition, error)
func (*Store) Reconcile ¶
Reconcile generates at most one due execution per Watch in one transaction. Missed intervals coalesce to the first evaluation after service resumes; a job keeps its stable ID over retries and restarts. Claiming is a separate operation.
func (*Store) ReconcileScope ¶ added in v0.115.0
ReconcileScope handles at most maxWatches due observations; queue wakeups can repeat it. Each Watch gets its own atomic transaction, avoiding a namespace snapshot and one transaction's budget growing with unrelated project history.
func (*Store) RegisterRunner ¶
func (s *Store) RegisterRunner(ctx context.Context, identity RunnerIdentity) error
func (*Store) RenewTelegramPoll ¶
func (s *Store) RenewTelegramPoll(ctx context.Context, id RunnerIdentity, fence int64) (*TelegramPollLease, error)
func (*Store) ResetBaseline ¶
func (s *Store) ResetBaseline(ctx context.Context, a Actor, id, operationID string, expectedVersion int64) (Definition, error)
ResetBaseline explicitly acknowledges a reviewed source revision. Cosmetic edits deliberately cannot clear the accepted semantic fingerprint.
func (*Store) RetryNotification ¶
RetryNotification retries an owned failed delivery using its original event and quota reservation. Revoked subscriptions are checked again before dispatch.
func (*Store) RevalidateSources ¶ added in v0.115.0
func (s *Store) RevalidateSources(ctx context.Context, scope Scope, maxWatches int, after string) (string, error)
RevalidateSources checks at most twenty project definitions after a durable document-ID cursor. Source-change consumers persist the returned continuation before acknowledging their producer event. An empty cursor means the bounded pass is complete. Transient admission errors abort without blocking a Watch.
func (*Store) SetLifecycle ¶
func (*Store) TelegramProviderStatus ¶
func (*Store) TelegramStatus ¶
func (*Store) WithClaim ¶ added in v0.115.0
func (s *Store) WithClaim(ctx context.Context, id RunnerIdentity, runID string, fence int64, fn func(context.Context, dal.ReadTransaction, Claim) error) error
WithClaim admits and reads the immutable source bundle in the exact same authorized snapshot as the current fenced lease. Callback results are usable only when this method returns nil; callbacks must not perform external work.
type Subscription ¶
type Subscription struct {
TelegramBotID string `json:"telegramBotId,omitempty"`
AccountID string `json:"accountId,omitempty"`
Tier string `json:"tier,omitempty"`
Phone string `json:"phone,omitempty"`
WatchID string `json:"watchId"`
UserID string `json:"userId"`
Channels []string `json:"channels"`
NotifyOn []string `json:"notifyOn,omitempty"`
Email string `json:"email,omitempty"`
Unsubscribed bool `json:"unsubscribed,omitempty"`
}
type TelegramPollLease ¶
type TelegramPollLease struct {
NextAttemptAt time.Time `json:"nextAttemptAt,omitempty"`
BotID string `json:"botId"`
Fence int64 `json:"fence"`
Offset int64 `json:"offset"`
LeaseUntil time.Time `json:"leaseUntil"`
}
TelegramPollLease fences a single provider poller and its persisted update offset. Offset is advanced only after an update has been handled.
type TelegramProviderStatus ¶
type TelegramProviderStatus struct {
ReconnectRequired bool `json:"reconnectRequired,omitempty"`
ConnectionError string `json:"connectionError,omitempty"`
Connected bool `json:"connected"`
BotID string `json:"botId,omitempty"`
BotUsername string `json:"botUsername,omitempty"`
}
TelegramProviderStatus exposes only the registered public bot identity and the requesting user's verified link. Bot secrets remain on provider runners.
Source Files
¶
- definitions.go
- events.go
- firestore.go
- firestore_admission.go
- firestore_changes.go
- firestore_claim.go
- firestore_events.go
- firestore_fence.go
- firestore_inbox.go
- firestore_operations.go
- firestore_query.go
- firestore_retention.go
- firestore_session.go
- model.go
- notifications.go
- retention.go
- scheduler.go
- storage.go
- telegram.go
- telegram_provider.go
- validate.go