convoy

README

https://getconvoy.io

Convoy is a fast & secure webhooks service. It receives event data from a HTTP API and sends these event data to the configured endpoints. To get started download the openapi spec into Postman or Insomnia.

Installation

You can either download the docker image or compile from source

Docker

docker pull ghcr.io/frain-dev/convoy:v0.1.0

Compile from source

git clone https://github.com/frain-dev/convoy.git
cd convoy
go build -o convoy ./cmd

Concepts

1. Apps: An app is an abstraction representing a user who wants to receive webhooks. Currently, an app contains one endpoint to receive webhooks.
2. Events: An event represents a webhook event to be sent to an app.
3. Delivery Attempts: A delivery attempt represents an attempt to send an event to it's respective app's endpoint. It contains the event body, status code and response body received on attempt. The amount of attempts on a failed delivery depends on your configured retry strategy.

Configuration

Convoy is configured using a json file with a sample configuration below:

{
	"database": {
		"dsn": "mongo-url-with-username-and-password"
	},
	"queue": {
		"type": "redis",
		"redis": {
			"dsn": "redis-url-with-username-and-password"
		}
	},
	"server": {
		"http": {
			"port": 5005
		}
	},
	"auth": {
		"type": "none"
	},
	"strategy": {
		"type": "default",
		"default": {
			"intervalSeconds": 125,
			"retryLimit": 15
		}
	},
	"signature": {
		"header": "X-Company-Event-Webhook-Signature"
	}
}

Notes to Configuration

• You can set basic auth mechanism with the following:

{
	"auth": {
		"type": "basic",
		"basic": {
			"username": "username",
			"password": "password"
		}
	}
}

Documentation

Index

• Variables
• func IsValidPeriod(period string) bool
• type AppMetadata
• type Application
• type ApplicationRepository
• type Datastore
• type DocumentStatus
• type Endpoint
• type EndpointMetadata
• type EndpointStatus
• type EventType
• type Group
  • func (o *Group) IsDeleted() bool
  • func (o *Group) IsOwner(a *Application) bool
• type GroupFilter
• type GroupRepository
• type HttpHeader
• type HttpMethod
• type Message
• type MessageAttempt
• type MessageMetadata
  • func (m MessageMetadata) Value() (driver.Value, error)
• type MessageRepository
• type MessageStatus
• type Period
• type Plugin
• type SentryHook
  • func NewSentryHook(levels []log.Level) *SentryHook
  • func (s *SentryHook) Fire(entry *log.Entry) error
  • func (s *SentryHook) Levels() []log.Level
• type TaskName

Variables

var (
	ErrApplicationNotFound = errors.New("application not found")

	ErrEndpointNotFound = errors.New("endpoint not found")
)

var (
	ErrMessageNotFound = errors.New("event not found")

	ErrMessageDeliveryAttemptNotFound = errors.New("delivery attempt not found")
)

var DefaultLevels = []log.Level{
	log.ErrorLevel,
	log.PanicLevel,
	log.FatalLevel,
	log.WarnLevel,
}

var ErrGroupNotFound = errors.New("group not found")

var PeriodValues = map[string]Period{
	"daily":   Daily,
	"weekly":  Weekly,
	"monthly": Monthly,
	"yearly":  Yearly,
}

Functions

func IsValidPeriod

func IsValidPeriod(period string) bool

Types

type AppMetadata

type AppMetadata struct {
	GroupID      string `json:"group_id" bson:"group_id"`
	Secret       string `json:"secret" bson:"secret"`
	SupportEmail string `json:"support_email" bson:"support_email"`

	Endpoints []EndpointMetadata `json:"endpoints" bson:"endpoints"`
}

type Application

type Application struct {
	ID           primitive.ObjectID `json:"-" bson:"_id"`
	UID          string             `json:"uid" bson:"uid"`
	GroupID      string             `json:"group_id" bson:"group_id"`
	Title        string             `json:"name" bson:"title"`
	SupportEmail string             `json:"support_email" bson:"support_email"`

	Secret string `json:"secret" bson:"secret"`

	Endpoints []Endpoint         `json:"endpoints" bson:"endpoints"`
	CreatedAt primitive.DateTime `json:"created_at,omitempty" bson:"created_at,omitempty"`
	UpdatedAt primitive.DateTime `json:"updated_at,omitempty" bson:"updated_at,omitempty"`
	DeletedAt primitive.DateTime `json:"deleted_at,omitempty" bson:"deleted_at,omitempty"`

	Events int64 `json:"events" bson:"-"`

	DocumentStatus DocumentStatus `json:"-" bson:"document_status"`
}

type ApplicationRepository

type ApplicationRepository interface {
	CreateApplication(context.Context, *Application) error
	LoadApplicationsPaged(context.Context, string, models.Pageable) ([]Application, pager.PaginationData, error)
	FindApplicationByID(context.Context, string) (*Application, error)
	UpdateApplication(context.Context, *Application) error
	DeleteApplication(context.Context, *Application) error
	LoadApplicationsPagedByGroupId(context.Context, string, models.Pageable) ([]Application, pager.PaginationData, error)
	SearchApplicationsByGroupId(context.Context, string, models.SearchParams) ([]Application, error)
	FindApplicationEndpointByID(context.Context, string, string) (*Endpoint, error)
	UpdateApplicationEndpointsStatus(context.Context, string, []string, EndpointStatus) error
}

type Datastore

type Datastore interface {
	GroupRepository
	ApplicationRepository
	// EndpointRepository
	io.Closer
	Migrate() error
}

Datastore provides an abstraction for all database related operations

type DocumentStatus

type DocumentStatus string

const (
	ActiveDocumentStatus   DocumentStatus = "Active"
	InactiveDocumentStatus DocumentStatus = "Inactive"
	DeletedDocumentStatus  DocumentStatus = "Deleted"
)

type Endpoint

type Endpoint struct {
	UID         string         `json:"uid" bson:"uid"`
	TargetURL   string         `json:"target_url" bson:"target_url"`
	Description string         `json:"description" bson:"description"`
	Status      EndpointStatus `json:"status" bson:"status"`

	CreatedAt primitive.DateTime `json:"created_at,omitempty" bson:"created_at,omitempty"`
	UpdatedAt primitive.DateTime `json:"updated_at,omitempty" bson:"updated_at,omitempty"`
	DeletedAt primitive.DateTime `json:"deleted_at,omitempty" bson:"deleted_at,omitempty"`

	DocumentStatus DocumentStatus `json:"-" bson:"document_status"`
}

type EndpointMetadata

type EndpointMetadata struct {
	UID       string         `json:"uid" bson:"uid"`
	TargetURL string         `json:"target_url" bson:"target_url"`
	Status    EndpointStatus `json:"status" bson:"status"`

	Sent bool `json:"sent" bson:"sent"`
}

type EndpointStatus

type EndpointStatus string

const (
	ActiveEndpointStatus   EndpointStatus = "active"
	InactiveEndpointStatus EndpointStatus = "inactive"
	PendingEndpointStatus  EndpointStatus = "pending"
)

type EventType

type EventType string

EventType is used to identify an specific event. This could be "user.new" This will be used for data indexing Makes it easy to filter by a list of events

type Group

type Group struct {
	ID      primitive.ObjectID `json:"-" bson:"_id"`
	UID     string             `json:"uid" bson:"uid"`
	Name    string             `json:"name" bson:"name"`
	LogoURL string             `json:"logo_url" bson:"logo_url"`

	CreatedAt primitive.DateTime `json:"created_at,omitempty" bson:"created_at,omitempty"`
	UpdatedAt primitive.DateTime `json:"updated_at,omitempty" bson:"updated_at,omitempty"`
	DeletedAt primitive.DateTime `json:"deleted_at,omitempty" bson:"deleted_at,omitempty"`

	DocumentStatus DocumentStatus `json:"-" bson:"document_status"`
}

func (*Group) IsDeleted

func (o *Group) IsDeleted() bool

func (*Group) IsOwner

func (o *Group) IsOwner(a *Application) bool

type GroupFilter

type GroupFilter struct {
	Name string `json:"name" bson:"name"`
}

type GroupRepository

type GroupRepository interface {
	LoadGroups(context.Context, *GroupFilter) ([]*Group, error)
	CreateGroup(context.Context, *Group) error
	UpdateGroup(context.Context, *Group) error
	FetchGroupByID(context.Context, string) (*Group, error)
}

type HttpHeader

type HttpHeader map[string]string

type HttpMethod

type HttpMethod string

const (
	HttpPost HttpMethod = "POST"
)

type Message

type Message struct {
	ID        primitive.ObjectID `json:"-" bson:"_id"`
	UID       string             `json:"uid" bson:"uid"`
	AppID     string             `json:"app_id" bson:"app_id"`
	EventType EventType          `json:"event_type" bson:"event_type"`

	// ProviderID is a custom ID that can be used to reconcile this message
	// with your internal systems.
	// This is optional
	// If not provided, we will generate one for you
	ProviderID string `json:"provider_id" bson:"provider_id"`

	// Data is an arbitrary JSON value that gets sent as the body of the
	// webhook to the endpoints
	Data json.RawMessage `json:"data" bson:"data"`

	Metadata *MessageMetadata `json:"metadata" bson:"metadata"`

	Description string `json:"description,omitempty" bson:"description"`

	Status MessageStatus `json:"status" bson:"status"`

	AppMetadata *AppMetadata `json:"app_metadata,omitempty" bson:"app_metadata"`

	MessageAttempts []MessageAttempt `json:"-" bson:"attempts"`

	CreatedAt primitive.DateTime `json:"created_at,omitempty" bson:"created_at,omitempty"`
	UpdatedAt primitive.DateTime `json:"updated_at,omitempty" bson:"updated_at,omitempty"`
	DeletedAt primitive.DateTime `json:"deleted_at,omitempty" bson:"deleted_at,omitempty"`

	DocumentStatus DocumentStatus `json:"-" bson:"document_status"`
}

Message defines a payload to be sent to an application

type MessageAttempt

type MessageAttempt struct {
	ID         primitive.ObjectID `json:"-" bson:"_id"`
	UID        string             `json:"uid" bson:"uid"`
	MsgID      string             `json:"msg_id" bson:"msg_id"`
	URL        string             `json:"url" bson:"url"`
	Method     string             `json:"method" bson:"method"`
	EndpointID string             `json:"endpoint_id" bson:"endpoint_id"`
	APIVersion string             `json:"api_version" bson:"api_version"`

	IPAddress        string        `json:"ip_address,omitempty" bson:"ip_address,omitempty"`
	RequestHeader    HttpHeader    `json:"request_http_header,omitempty" bson:"request_http_header,omitempty"`
	ResponseHeader   HttpHeader    `json:"response_http_header,omitempty" bson:"response_http_header,omitempty"`
	HttpResponseCode string        `json:"http_status,omitempty" bson:"http_status,omitempty"`
	ResponseData     string        `json:"response_data,omitempty" bson:"response_data,omitempty"`
	Error            string        `json:"error,omitempty" bson:"error,omitempty"`
	Status           MessageStatus `json:"status,omitempty" bson:"status,omitempty"`

	CreatedAt primitive.DateTime `json:"created_at,omitempty" bson:"created_at,omitempty"`
	UpdatedAt primitive.DateTime `json:"updated_at,omitempty" bson:"updated_at,omitempty"`
	DeletedAt primitive.DateTime `json:"deleted_at,omitempty" bson:"deleted_at,omitempty"`
}

type MessageMetadata

type MessageMetadata struct {
	Strategy config.StrategyProvider `json:"strategy" bson:"strategy"`
	// NextSendTime denotes the next time a message will be published in
	// case it failed the first time
	NextSendTime primitive.DateTime `json:"next_send_time" bson:"next_send_time"`

	// NumTrials: number of times we have tried to deliver this message to
	// an application
	NumTrials uint64 `json:"num_trials" bson:"num_trials"`

	IntervalSeconds uint64 `json:"interval_seconds" bson:"interval_seconds"`

	RetryLimit uint64 `json:"retry_limit" bson:"retry_limit"`
}

func (MessageMetadata) Value

func (m MessageMetadata) Value() (driver.Value, error)

type MessageRepository

type MessageRepository interface {
	CreateMessage(context.Context, *Message) error
	LoadMessageIntervals(context.Context, string, models.SearchParams, Period, int) ([]models.MessageInterval, error)
	LoadMessagesPagedByAppId(context.Context, string, models.SearchParams, models.Pageable) ([]Message, pager.PaginationData, error)
	FindMessageByID(ctx context.Context, id string) (*Message, error)
	LoadMessagesScheduledForPosting(context.Context) ([]Message, error)
	LoadMessagesForPostingRetry(context.Context) ([]Message, error)
	LoadAbandonedMessagesForPostingRetry(context.Context) ([]Message, error)
	UpdateStatusOfMessages(context.Context, []Message, MessageStatus) error
	UpdateMessageWithAttempt(ctx context.Context, m Message, attempt MessageAttempt) error
	LoadMessagesPaged(context.Context, string, string, models.SearchParams, models.Pageable) ([]Message, pager.PaginationData, error)
}

type MessageStatus

type MessageStatus string

const (
	// ScheduledMessageStatus : when  a message has been scheduled for delivery
	ScheduledMessageStatus  MessageStatus = "Scheduled"
	ProcessingMessageStatus MessageStatus = "Processing"
	DiscardedMessageStatus  MessageStatus = "Discarded"
	FailureMessageStatus    MessageStatus = "Failure"
	SuccessMessageStatus    MessageStatus = "Success"
	RetryMessageStatus      MessageStatus = "Retry"
)

type Period

type Period int

const (
	Daily Period = iota
	Weekly
	Monthly
	Yearly
)

type Plugin

type Plugin interface {
	Apply(http.ResponseWriter, *http.Request) error
	Name() string
	IsEnabled() bool
}

type SentryHook

type SentryHook struct {
	LogLevels []log.Level
}

func NewSentryHook

func NewSentryHook(levels []log.Level) *SentryHook

func (*SentryHook) Fire

func (s *SentryHook) Fire(entry *log.Entry) error

func (*SentryHook) Levels

func (s *SentryHook) Levels() []log.Level

type TaskName

type TaskName string

const (
	EventProcessor      TaskName = "EventProcessor"
	DeadLetterProcessor TaskName = "DeadLetterProcessor"
)