Log Event Trigger Capability Development: Part 1

core/capabilities/log_event_trigger/log_event_trigger.go

@@ -0,0 +1,99 @@
+package log_event_trigger
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/hashicorp/go-plugin"
+
+	"github.com/smartcontractkit/chainlink/v2/core/capabilities/log_event_trigger/trigger"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	"github.com/smartcontractkit/chainlink-common/pkg/loop"
+	"github.com/smartcontractkit/chainlink-common/pkg/types/core"
+)
+
+const (
+	serviceName = "LogEventTriggerCapability"
+)
+
+type LogEventServiceGRPC struct {
+	trigger capabilities.TriggerCapability
+	s       *loop.Server
+}
+
+func main() {
+	s := loop.MustNewStartedServer(serviceName)
+	defer s.Stop()
+
+	s.Logger.Infof("Starting %s", serviceName)
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+
+	plugin.Serve(&plugin.ServeConfig{
+		HandshakeConfig: loop.StandardCapabilitiesHandshakeConfig(),
+		Plugins: map[string]plugin.Plugin{
+			loop.PluginStandardCapabilitiesName: &loop.StandardCapabilitiesLoop{
+				PluginServer: &LogEventServiceGRPC{
+					s: s,
+				},
+				BrokerConfig: loop.BrokerConfig{Logger: s.Logger, StopCh: stopCh, GRPCOpts: s.GRPCOpts},
+			},
+		},
+		GRPCServer: s.GRPCOpts.NewServer,
+	})
+}
+
+func (cs *LogEventServiceGRPC) Start(ctx context.Context) error {
+	return nil
+}
+
+func (cs *LogEventServiceGRPC) Close() error {
+	return nil
+}
+
+func (cs *LogEventServiceGRPC) Ready() error {
+	return nil
+}
+
+func (cs *LogEventServiceGRPC) HealthReport() map[string]error {
+	return nil
+}
+
+func (cs *LogEventServiceGRPC) Name() string {
+	return serviceName
+}
+
+func (cs *LogEventServiceGRPC) Infos(ctx context.Context) ([]capabilities.CapabilityInfo, error) {
+	triggerInfo, err := cs.trigger.Info(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return []capabilities.CapabilityInfo{
+		triggerInfo,
+	}, nil
+}
+
+func (cs *LogEventServiceGRPC) Initialise(
+	ctx context.Context,
+	config string,
+	telemetryService core.TelemetryService,
+	store core.KeyValueStore,
+	capabilityRegistry core.CapabilitiesRegistry,
+	errorLog core.ErrorLog,
+	pipelineRunner core.PipelineRunnerService,
+	relayerSet core.RelayerSet,
+) error {
+	cs.s.Logger.Debugf("Initialising %s", serviceName)
+	cs.trigger = trigger.New(trigger.Params{
+		Logger: cs.s.Logger,
+	})
+
+	if err := capabilityRegistry.Add(ctx, cs.trigger); err != nil {
+		return fmt.Errorf("error when adding cron trigger to the registry: %w", err)
+	}
+
+	return nil
+}

core/capabilities/log_event_trigger/trigger/store.go

@@ -0,0 +1,68 @@
+package trigger
+
+import (
+	"fmt"
+	"sync"
+)
+
+type NewCapabilityFn[T any, Resp any] func() (T, chan Resp)
+
+// Interface of the capabilities store
+type CapabilitiesStore[T any, Resp any] interface {
+	Read(capabilityID string) (value T, ok bool)
+	ReadAll() (values map[string]T)
+	Write(capabilityID string, value T)
+	InsertIfNotExists(capabilityID string, fn NewCapabilityFn[T, Resp]) (chan Resp, error)
+	Delete(capabilityID string)
+}
+
+// Implementation for the CapabilitiesStore interface
+type capabilitiesStore[T any, Resp any] struct {
+	mu           sync.RWMutex
+	capabilities map[string]T
+}
+
+var _ CapabilitiesStore[string, string] = (CapabilitiesStore[string, string])(nil)
+
+// Constructor for capabilitiesStore struct implementing CapabilitiesStore interface
+func NewCapabilitiesStore[T any, Resp any]() CapabilitiesStore[T, Resp] {
+	return &capabilitiesStore[T, Resp]{
+		capabilities: map[string]T{},
+	}
+}
+
+func (cs *capabilitiesStore[T, Resp]) Read(capabilityID string) (value T, ok bool) {
+	cs.mu.RLock()
+	defer cs.mu.RUnlock()
+	trigger, ok := cs.capabilities[capabilityID]
+	return trigger, ok
+}
+
+func (cs *capabilitiesStore[T, Resp]) ReadAll() (values map[string]T) {
+	cs.mu.RLock()
+	defer cs.mu.RUnlock()
+	return cs.capabilities
+}
+
+func (cs *capabilitiesStore[T, Resp]) Write(capabilityID string, value T) {
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	cs.capabilities[capabilityID] = value
+}
+
+func (cs *capabilitiesStore[T, Resp]) InsertIfNotExists(capabilityID string, fn NewCapabilityFn[T, Resp]) (chan Resp, error) {
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	if _, ok := cs.capabilities[capabilityID]; ok {
+		return nil, fmt.Errorf("capabilityID %v already exists", capabilityID)
+	}
+	value, respCh := fn()
+	cs.capabilities[capabilityID] = value
+	return respCh, nil
+}
+
+func (cs *capabilitiesStore[T, Resp]) Delete(capabilityID string) {
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	delete(cs.capabilities, capabilityID)
+}

core/capabilities/log_event_trigger/trigger/trigger.go

@@ -0,0 +1,122 @@
+package trigger
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
+)
+
+const ID = "[email protected]"
+
+const defaultSendChannelBufferSize = 1000
+
+var logEventTriggerInfo = capabilities.MustNewCapabilityInfo(
+	ID,
+	capabilities.CapabilityTypeTrigger,
+	"A trigger that listens for specific contract log events and starts a workflow run.",
+)
+
+// Log Event Trigger Capability Config
+type Config struct {
+}
+
+// Log Event Trigger Capability Payload
+type Payload struct {
+	// Time that Log Event Trigger's task execution occurred (RFC3339Nano formatted)
+	ActualExecutionTime string
+}
+
+// Log Event Trigger Capability Response
+type Response struct {
+	capabilities.TriggerEvent
+	Metadata struct{}
+	Payload  Payload
+}
+
+type logEventTrigger struct {
+	ch chan<- capabilities.TriggerResponse
+}
+
+// Log Event Trigger Capabilities Manager
+// Manages different log event triggers using an underlying triggerStore
+type LogEventTriggerManager struct {
+	capabilities.CapabilityInfo
+	capabilities.Validator[Config, struct{}, capabilities.TriggerResponse]
+	lggr     logger.Logger
+	triggers CapabilitiesStore[logEventTrigger, capabilities.TriggerResponse]
+}
+
+type Params struct {
+	Logger logger.Logger
+}
+
+var _ capabilities.TriggerCapability = (*LogEventTriggerManager)(nil)
+var _ services.Service = &LogEventTriggerManager{}
+
+// Creates a new Cron Trigger Service.
+// Scheduling will commence on calling .Start()
+func New(p Params) *LogEventTriggerManager {
+	l := logger.Named(p.Logger, "Log Event Trigger Capability Service")
+
+	logEventStore := NewCapabilitiesStore[logEventTrigger, capabilities.TriggerResponse]()
+
+	return &LogEventTriggerManager{
+		CapabilityInfo: logEventTriggerInfo,
+		lggr:           l,
+		triggers:       logEventStore,
+	}
+}
+
+// Register a new trigger
+// Can register triggers before the service is actively scheduling
+func (s *LogEventTriggerManager) RegisterTrigger(ctx context.Context, req capabilities.TriggerRegistrationRequest) (<-chan capabilities.TriggerResponse, error) {
+	if req.Config == nil {
+		return nil, errors.New("config is required to register a log event trigger")
+	}
+	_, err := s.ValidateConfig(req.Config)
+	if err != nil {
+		return nil, err
+	}
+	respCh, err := s.triggers.InsertIfNotExists(req.TriggerID, func() (logEventTrigger, chan capabilities.TriggerResponse) {
+		callbackCh := make(chan capabilities.TriggerResponse, defaultSendChannelBufferSize)
+		return logEventTrigger{
+			ch: callbackCh,
+		}, callbackCh
+	})
+	if err != nil {
+		return nil, fmt.Errorf("log_event_trigger %v", err)
+	}
+	return respCh, nil
+}
+
+func (s *LogEventTriggerManager) UnregisterTrigger(ctx context.Context, req capabilities.TriggerRegistrationRequest) error {
+	return nil
+}
+
+// Start the service.
+func (s *LogEventTriggerManager) Start(ctx context.Context) error {
+	return nil
+}
+
+// Close stops the Service.
+// After this call the Service cannot be started again,
+// The service will need to be re-built to start scheduling again.
+func (s *LogEventTriggerManager) Close() error {
+	return nil
+}
+
+func (s *LogEventTriggerManager) Ready() error {
+	return nil
+}
+
+func (s *LogEventTriggerManager) HealthReport() map[string]error {
+	return map[string]error{s.Name(): nil}
+}
+
+func (s *LogEventTriggerManager) Name() string {
+	return "Service"
+}