watch

package
v0.114.0 Latest Latest
Warning

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

 Go to latest Published: Oct 24, 2026 License: Apache-2.0 Imports: 23 Imported by: 0

Documentation

Overview

Package watch implements project Watches over DALGO transactional state.

Index

Constants

This section is empty.

Variables

View Source 
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")
)

Functions

This section is empty.

Types

type Actor 

type Actor struct {
	ID           string
	Email        string
	Scope        Scope
	SharedEditor bool
	AccountID    string
	Tier         string
	Phone        string
}

Actor is supplied by the authenticated transport, never decoded from a request.

type AuditEvent 

type AuditEvent struct {
	ID         string    `json:"id"`
	Scope      Scope     `json:"scope"`
	Actor      string    `json:"actor"`
	Action     string    `json:"action"`
	ResourceID string    `json:"resourceId"`
	At         time.Time `json:"at"`
}

type Claim 

type Claim struct {
	ClaimID    string     `json:"claimId"`
	RunID      string     `json:"runId"`
	Fence      int64      `json:"fence"`
	LeaseUntil time.Time  `json:"leaseUntil"`
	Definition Definition `json:"definition"`
}

type Completion 

type Completion struct {
	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 {
	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 Notification 

type Notification struct {
	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 NotificationPolicy struct {
	CooldownSeconds int  `json:"cooldownSeconds,omitempty"`
	Recovery        bool `json:"recovery"`
}

type Observation 

type Observation struct {
	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 Run 

type Run struct {
	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 Runner 

type Runner struct {
	TelegramBotID       string `json:"telegramBotId,omitempty"`
	TelegramBotUsername string `json:"telegramBotUsername,omitempty"`
	Scope
	ID           string    `json:"id"`
	Capabilities []string  `json:"capabilities"`
	LastSeen     time.Time `json:"lastSeen"`
}

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 Schedule 

type Schedule struct {
	IntervalSeconds int `json:"intervalSeconds"`
}

type Scope 

type Scope struct {
	Namespace   string `json:"namespace"`
	Project     string `json:"project"`
	Environment string `json:"environment"`
}

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 NewSQLiteStore 

func NewSQLiteStore(connection string, c Config) (*Store, error)

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 (s *Store) AdvanceTelegramPoll(ctx context.Context, id RunnerIdentity, fence, offset int64) error

func (*Store) Audit 

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

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) ClaimDue 

func (s *Store) ClaimDue(ctx context.Context, identity RunnerIdentity) (*Claim, error)

func (*Store) ClaimNotifications 

func (s *Store) ClaimNotifications(ctx context.Context) (*Delivery, error)

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 (s *Store) ClaimTelegramNotification(ctx context.Context, id RunnerIdentity, pollFence int64) (*Delivery, error)

func (*Store) ClaimTelegramPoll 

func (s *Store) ClaimTelegramPoll(ctx context.Context, id RunnerIdentity) (*TelegramPollLease, error)

func (*Store) Close 

func (s *Store) Close() 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 (s *Store) DeleteDefinition(ctx context.Context, a Actor, id, operationID string, expectedVersion int64) error

func (*Store) DisconnectTelegram 

func (s *Store) DisconnectTelegram(ctx context.Context, a Actor) error

func (*Store) FinishNotification 

func (s *Store) FinishNotification(ctx context.Context, id string, fence int64, failureCode string) error

func (*Store) FinishRun 

func (s *Store) FinishRun(ctx context.Context, identity RunnerIdentity, c Completion) (Run, error)

func (*Store) FinishTelegramNotification 

func (s *Store) FinishTelegramNotification(ctx context.Context, identity RunnerIdentity, pollFence int64, id string, fence int64, failureCode string, retryAfterSeconds ...int) error

func (*Store) GetDefinition 

func (s *Store) GetDefinition(ctx context.Context, a Actor, id string) (Definition, error)

func (*Store) GetSubscription 

func (s *Store) GetSubscription(ctx context.Context, a Actor, watchID string) (Subscription, error)

func (*Store) Health 

func (s *Store) Health(ctx context.Context) error

Health reads persisted storage, so an unavailable file is not reported healthy.

func (*Store) Heartbeat 

func (s *Store) Heartbeat(ctx context.Context, identity RunnerIdentity, id string, fence int64) error

func (*Store) Inbox 

func (s *Store) Inbox(ctx context.Context, a Actor) ([]Notification, error)

func (*Store) ListDefinitions 

func (s *Store) ListDefinitions(ctx context.Context, a Actor) ([]Definition, error)

func (*Store) ListDeliveries 

func (s *Store) ListDeliveries(ctx context.Context, a Actor, watchID string) ([]Delivery, error)

func (*Store) ListRuns 

func (s *Store) ListRuns(ctx context.Context, a Actor, watchID string) ([]Run, error)

func (*Store) MarkRead 

func (s *Store) MarkRead(ctx context.Context, a Actor, id string) error

func (*Store) PutDefinition 

func (s *Store) PutDefinition(ctx context.Context, a Actor, d Definition, operationID string, expectedVersion int64) (Definition, error)

func (*Store) Reconcile 

func (s *Store) Reconcile(ctx context.Context) error

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) 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 

func (s *Store) RetryNotification(ctx context.Context, a Actor, id string) error

RetryNotification retries an owned failed delivery using its original event and quota reservation. Revoked subscriptions are checked again before dispatch.

func (*Store) SetLifecycle 

func (s *Store) SetLifecycle(ctx context.Context, a Actor, id, lifecycle, operationID string, expectedVersion int64) (Definition, error)

func (*Store) Status 

func (s *Store) Status(ctx context.Context, a Actor) (Status, error)

func (*Store) Subscribe 

func (s *Store) Subscribe(ctx context.Context, a Actor, sub Subscription) error

func (*Store) TelegramProviderStatus 

func (s *Store) TelegramProviderStatus(ctx context.Context, a Actor) (TelegramProviderStatus, error)

func (*Store) TelegramStatus 

func (s *Store) TelegramStatus(ctx context.Context, a Actor) (bool, error)

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

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.