mongowatch

README

mongowatch

MongoDB Event Stream Watcher

Watches target mongo collection using mongo event log and executes certain handlers based on subscribed collection changes.

Self healing

If the collection was renamed, dropped or recreated, the event stream produces an 'invalidate' event for which the watcher is implemented to recover automatically from.

However make sure to reapply the collMod command options to the collection (if necessary).

This package contains helper methods to do it (make sure you have the right Mongo user permissions):

For Mongo >= 6.0

collMod changeStreamPreAndPostImages

db.EnablePrePostImages(mongoInstance *mongo.Database, colName string) error

For Mongo < 6.0

collMod recordPreImages

db.RecordPreImages(mongoInstance *mongo.Database, colName string) error

Package testing

To be able to run tests in this repo you will need to have some local and remote mongo instances running on port 27017. Configure parts with TODO comments.

Courtesy of @ignasbernotas and @zolia

Documentation

Index

• Constants
• type ChangeEventDispatcherFunc
• type ChangeStreamEvent
• type ChangeStreamResumePoint
• type ChangeStreamWatcher
• type CollectionWatcher
• type DocumentProcessor
• type ResumeToken
• type StreamResume

Constants

const OperationTypeInvalidate = "invalidate"

Types

type ChangeEventDispatcherFunc

type ChangeEventDispatcherFunc func(ctx context.Context, ce ChangeStreamEvent, err error) error

ChangeEventDispatcherFunc change event callback function returning err will stop further ChangeEventDispatcherFunc processing and the change stream watcher

type ChangeStreamEvent

type ChangeStreamEvent struct {
	ID            ResumeToken         `bson:"_id" json:"_id"`
	User          string              `bson:"user" json:"user"`
	Timestamp     primitive.Timestamp `bson:"timestamp" json:"timestamp"`
	OperationType string              `bson:"operationType" json:"operationType"`
	Database      string              `bson:"database" json:"database"`
	Collection    string              `bson:"collection" json:"collection"`
	// DocumentKey is the unique identifier for the document that was changed
	// (e.g. the _id field for a document)
	// some of our collections use custom IDs therefore it doesn't fit into the primitive.ObjectID type
	DocumentKey              string      `bson:"documentKey" json:"documentKey"`
	FullDocument             primitive.M `bson:"fullDocument" json:"fullDocument"`
	FullDocumentBeforeChange primitive.M `bson:"fullDocumentBeforeChange" json:"fullDocumentBeforeChange"`
	// TODO: get previous field values e.g. paidUntil
	UpdateDescription struct {
		UpdatedFields map[string]interface{} `bson:"updatedFields" json:"updatedFields"`
		RemovedFields interface{}            `bson:"removedFields" json:"removedFields"`
	} `bson:"updateDescription" json:"updateDescription"`
}

ChangeStreamEvent is the customized representation of a MongoDB change stream event that is captured and processed by this application.

type ChangeStreamResumePoint

type ChangeStreamResumePoint struct {
	ID        ResumeToken         `bson:"_id" json:"_id"`
	Timestamp primitive.Timestamp `bson:"timestamp" json:"timestamp"`
	// need to keep this for tests
	FullDocument primitive.M `bson:"fullDocument" json:"fullDocument"`
	// important to know before resuming the stream
	// OperationType == 'invalidate' means that the resume point is no longer valid,
	// and we need to use startAfter to resume the stream
	OperationType string `bson:"operationType" json:"operationType"`
}

ChangeStreamResumePoint holds information needed to resume a change stream from a specific point

type ChangeStreamWatcher

type ChangeStreamWatcher interface {
	// Start resumes watching change events and
	// passes event data to the supplied dispatch function for handling
	Start(ctx context.Context, fullDocumentMode options.FullDocument, resumePoint *ChangeStreamResumePoint, saveFunc, deleteFunc ChangeEventDispatcherFunc, dispatchFuncs ...ChangeEventDispatcherFunc) error
}

ChangeStreamWatcher watches a change stream and dispatches received changed events

type CollectionWatcher

type CollectionWatcher interface {
	Update(ctx context.Context, doc []byte) error
	Insert(ctx context.Context, doc []byte) error
	Delete(ctx context.Context, doc []byte) error
}

CollectionWatcher is an interface for processing document data from a change stream

type DocumentProcessor

type DocumentProcessor interface {
	StartWithRetry(bo backoff.BackOff, actions CollectionWatcher, fullDocumentMode options.FullDocument) error
	Start(actions CollectionWatcher, fullDocumentMode options.FullDocument) error
	Stop()
}

DocumentProcessor is an interface for processing document data from a change stream

type ResumeToken

type ResumeToken struct {
	TokenData interface{} `bson:"_data" json:"_data"`
}

ResumeToken denotes the token associated with a MongoDB change stream event, which may be used to resume receiving change stream events from a point in the past.

type StreamResume

type StreamResume interface {
	// GetResumePoint fetches the last stored resume point
	GetResumePoint() (*ChangeStreamResumePoint, error)
	// GetResumeTime fetches the last stored resume point and extracts the timestamp
	GetResumeTime() (*primitive.Timestamp, error)
	// DeleteResumePoint deletes a change stream resume point from the collection
	DeleteResumePoint(ctx context.Context, token ResumeToken) error
	// SaveResumePoint stores ChangeStreamResumePoint
	SaveResumePoint(ctx context.Context, ce ChangeStreamResumePoint) error
}

StreamResume stores relevant change stream events mongo's oplog has configurable expiration, but we don't need a large oplog instead we store the changes we actually need