sync

Documentation

Index

• type EventBatch
• type GossipOptions
• type Identity
• type InboxEventResult
• type InboxResponse
• type Logger
• type Options
• type OutboxPage
• type Result
• type Service
  • func New(identity Identity, store Store, logger Logger) *Service
  • func NewWithOptions(identity Identity, store Store, logger Logger, opts Options) *Service
  • func (s *Service) GossipOnce(ctx context.Context, opts GossipOptions) (result Result, err error)
  • func (s *Service) PushOnce(ctx context.Context, limit int) (result Result, err error)
  • func (s *Service) RetryPendingProjections(ctx context.Context, limit int) (int, error)
  • func (s *Service) Run(ctx context.Context, interval time.Duration) error
  • func (s *Service) RunGossip(ctx context.Context, interval time.Duration, opts GossipOptions) error
  • func (s *Service) RunPush(ctx context.Context, interval time.Duration, limit int) error
  • func (s *Service) SyncOnce(ctx context.Context) (result Result, err error)
  • func (s *Service) UpsertManualPeers(ctx context.Context, peerURLs []string) error
• type Store

Types

type EventBatch

type EventBatch struct {
	SchemaVersion string                `json:"schema_version"`
	Type          string                `json:"type"`
	Events        []*events.SignedEvent `json:"events"`
}

type GossipOptions

type GossipOptions struct {
	NetworkID           string
	TTL                 int
	BatchSize           int
	Fanout              int
	PeerExchangeEnabled bool
}

type Identity

type Identity interface {
	events.Identity
	PublicKey(context.Context) (ed25519.PublicKey, error)
}

type InboxEventResult

type InboxEventResult struct {
	EventID string `json:"event_id"`
	Status  string `json:"status"`
	Code    string `json:"code,omitempty"`
	Message string `json:"message,omitempty"`
}

type InboxResponse

type InboxResponse struct {
	SchemaVersion string             `json:"schema_version"`
	Type          string             `json:"type"`
	Accepted      []InboxEventResult `json:"accepted"`
	Duplicate     []InboxEventResult `json:"duplicate"`
	Rejected      []InboxEventResult `json:"rejected"`
	Cursor        string             `json:"cursor,omitempty"`
}

type Logger

type Logger interface {
	Info(format string, args ...any)
	Warn(format string, args ...any)
	Error(format string, args ...any)
}

type Options

type Options struct {
	AllowInsecurePeerHTTP bool
	EventTimeTolerance    time.Duration
	MaxEventAge           time.Duration
}

type OutboxPage

type OutboxPage struct {
	SchemaVersion string               `json:"schema_version"`
	Type          string               `json:"type"`
	Events        []events.SignedEvent `json:"events"`
	NextCursor    string               `json:"next_cursor"`
	HasMore       bool                 `json:"has_more"`
}

type Result

type Result struct {
	Peers     int `json:"peers"`
	Accepted  int `json:"accepted"`
	Duplicate int `json:"duplicate"`
	Rejected  int `json:"rejected"`
	Projected int `json:"projected"`
}

type Service

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

func New

func New(identity Identity, store Store, logger Logger) *Service

func NewWithOptions

func NewWithOptions(identity Identity, store Store, logger Logger, opts Options) *Service

func (*Service) GossipOnce

func (s *Service) GossipOnce(ctx context.Context, opts GossipOptions) (result Result, err error)

func (*Service) PushOnce

func (s *Service) PushOnce(ctx context.Context, limit int) (result Result, err error)

func (*Service) RetryPendingProjections

func (s *Service) RetryPendingProjections(ctx context.Context, limit int) (int, error)

RetryPendingProjections replays durable projection failures against the immutable accepted event log. It is safe to call repeatedly.

func (*Service) Run

func (s *Service) Run(ctx context.Context, interval time.Duration) error

func (*Service) RunGossip

func (s *Service) RunGossip(ctx context.Context, interval time.Duration, opts GossipOptions) error

func (*Service) RunPush

func (s *Service) RunPush(ctx context.Context, interval time.Duration, limit int) error

func (*Service) SyncOnce

func (s *Service) SyncOnce(ctx context.Context) (result Result, err error)

func (*Service) UpsertManualPeers

func (s *Service) UpsertManualPeers(ctx context.Context, peerURLs []string) error

type Store

type Store interface {
	UpsertFederationPeerURL(ctx context.Context, peerURL string) (int64, error)
	ListEnabledFederationPeers(ctx context.Context) ([]pgindex.FederationPeerRecord, error)
	UpsertFederationNode(ctx context.Context, node pgindex.FederationNodeRecord) error
	AppendVerifiedFederationEvent(ctx context.Context, event *events.SignedEvent, validation *events.ValidationResult) error
	AppendRejectedFederationEvent(ctx context.Context, eventID, authorNodeID, eventType string, rawEventJSON []byte, reason string) error
	UpsertFederatedReleaseCardProjection(ctx context.Context, projection releasecard.Projection) error
	ProjectValidatorCapacity(ctx context.Context, projection pgindex.ValidatorCapacityProjection) error
	ProjectArticleAvailabilityAttestation(ctx context.Context, projection pgindex.ArticleAvailabilityProjection) error
	ProjectChecksumAttestation(ctx context.Context, projection pgindex.ChecksumAttestationProjection) error
	ProjectManifestAvailability(ctx context.Context, projection pgindex.ManifestAvailabilityProjection) error
	ProjectTrustAttestation(ctx context.Context, projection pgindex.TrustAttestationProjection) error
	ProjectHealthAttestation(ctx context.Context, projection pgindex.HealthAttestationProjection) error
	ProjectTombstone(ctx context.Context, projection pgindex.TombstoneProjection) error
	ProjectCoverageEvent(ctx context.Context, event *events.SignedEvent) error
	MarkFederationPeerSyncSuccess(ctx context.Context, peerID int64, nodeID, cursor, lastEventID string) error
	MarkFederationPeerSyncFailure(ctx context.Context, peerID int64, errText string) error
	ListUndeliveredFederationEvents(ctx context.Context, peerID int64, nodeID string, limit int) ([]*events.SignedEvent, error)
	RecordFederationPeerDelivery(ctx context.Context, result pgindex.FederationDeliveryResult) error
	ValidateFederationPoolControlEvent(ctx context.Context, event *events.SignedEvent) error
	ProjectFederationPoolEvent(ctx context.Context, event *events.SignedEvent) error
	CanAcceptFederationEventForPools(ctx context.Context, authorNodeID string, poolIDs []string, eventType string) (pgindex.PoolAuthorizationResult, error)
	IsActivePoolMember(ctx context.Context, poolID, nodeID string) (bool, error)
}