convoy

README

Convoy

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 EventType
• type HttpMethod
• type Message
• type MessageAttempt
• type MessageMetadata
  • func (m MessageMetadata) Value() (driver.Value, error)
• type MessageRepository
• type MessageStatus
• type Organisation
  • func (o *Organisation) IsDeleted() bool
  • func (o *Organisation) IsOwner(a *Application) bool
• type OrganisationRepository
• type Period
• type Plugin

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 ErrOrganisationNotFound = errors.New("organisation 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 {
	OrgID  string `json:"org_id" bson:"org_id"`
	Secret string `json:"secret" bson:"secret"`

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

type Application

type Application struct {
	ID    primitive.ObjectID `json:"-" bson:"_id"`
	UID   string             `json:"uid" bson:"uid"`
	OrgID string             `json:"org_id" bson:"org_id"`
	Title string             `json:"name" bson:"title"`

	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
	LoadApplications(context.Context, string) ([]Application, error)
	FindApplicationByID(context.Context, string) (*Application, error)
	UpdateApplication(context.Context, *Application) error
	DeleteApplication(context.Context, *Application) error
	LoadApplicationsPagedByOrgId(context.Context, string, models.Pageable) ([]Application, pager.PaginationData, error)
	SearchApplicationsByOrgId(context.Context, string, models.SearchParams) ([]Application, error)
}

type Datastore

type Datastore interface {
	OrganisationRepository
	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"`

	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"`

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

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 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"`
	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"`
	ContentType      string        `json:"content_type,omitempty" bson:"content_type,omitempty"`
	Header           http.Header   `json:"http_header,omitempty" bson:"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.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, 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"
	FailureMessageStatus    MessageStatus = "Failure"
	SuccessMessageStatus    MessageStatus = "Success"
	RetryMessageStatus      MessageStatus = "Retry"
)

type Organisation

type Organisation struct {
	ID      primitive.ObjectID `json:"-" bson:"_id"`
	UID     string             `json:"uid" bson:"uid"`
	OrgName string             `json:"name" bson:"org_name"`

	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 (*Organisation) IsDeleted

func (o *Organisation) IsDeleted() bool

func (*Organisation) IsOwner

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

type OrganisationRepository

type OrganisationRepository interface {
	LoadOrganisations(context.Context) ([]*Organisation, error)
	CreateOrganisation(context.Context, *Organisation) error
	UpdateOrganisation(context.Context, *Organisation) error
	FetchOrganisationByID(context.Context, string) (*Organisation, error)
}

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
}