postgresworker

README

Example

This is example for create Postgres Event Listener (CDC), inspired by Hasura Event Triggers

Create delivery handler

package workerhandler

import (
	"context"
	"encoding/json"
	
	"example.service/internal/modules/examplemodule/delivery/workerhandler"

	"github.com/golangid/candi/candishared"
	"github.com/golangid/candi/codebase/factory/types"
	"github.com/golangid/candi/tracer"
)

// PostgresListenerHandler struct
type PostgresListenerHandler struct {
	uc        usecase.Usecase
	validator interfaces.Validator
}

// NewPostgresListenerHandler constructor
func NewPostgresListenerHandler(uc usecase.Usecase, validator interfaces.Validator) *PostgresListenerHandler {
	return &PostgresListenerHandler{
		uc:        uc,
		validator: validator,
	}
}

// MountHandlers mount handler group
func (h *PostgresListenerHandler) MountHandlers(group *types.WorkerHandlerGroup) {
	group.Add("table-names", h.handleDataChange) // listen change data capture on table "table-names"
}

func (h *PostgresListenerHandler) handleDataChange(eventContext *candishared.EventContext) error {
	trace := tracer.StartTrace(eventContext.Context(), "DeliveryPostgresListener:HandleDataChange")
	defer trace.Finish()

	fmt.Printf("data change on table 'table-names' detected: %s\n", eventContext.Message())
	// call usecase
	return nil
}

Register in module

package examplemodule

import (
	"example.service/internal/modules/examplemodule/delivery/workerhandler"

	"github.com/golangid/candi/codebase/factory/dependency"
	"github.com/golangid/candi/codebase/factory/types"
	"github.com/golangid/candi/codebase/interfaces"
)

type Module struct {
	// ...another delivery handler
	workerHandlers map[types.Worker]interfaces.WorkerHandler
}

func NewModules(deps dependency.Dependency) *Module {
	return &Module{
		workerHandlers: map[types.Worker]interfaces.WorkerHandler{
			// ...another worker handler
			// ...
			types.PostgresListener: workerhandler.NewPostgresListenerHandler(usecaseUOW.User(), deps.GetValidator()),
		},
	}
}

// ...another method

Message in JSON Payload

{
	"event_id": "<md5-hash-unique-event-id>",
	"table": "<table-name>",
	"action": "<operation-name>", // INSERT, UPDATE, or DELETE
	"data": {
		"old": <old-column-values-object>,
		"new": <new-column-values-object>
	}
}

Documentation

Index

• Constants
• func NewWorker(service factory.ServiceFactory, opts ...OptionFunc) factory.AppServerFactory
• type EventPayload
  • func (e EventPayload) GetID() string
• type EventPayloadData
• type OptionFunc
  • func SetDebugMode(debugMode bool) OptionFunc
  • func SetLocker(locker candiutils.Locker) OptionFunc
  • func SetMaxGoroutines(maxGoroutines int) OptionFunc
  • func SetPostgresDSN(dsn string) OptionFunc

Constants

const (
	// ActionInsert const
	ActionInsert = "INSERT"
	// ActionUpdate const
	ActionUpdate = "UPDATE"
	// ActionDelete const
	ActionDelete = "DELETE"
)

Functions

func NewWorker

func NewWorker(service factory.ServiceFactory, opts ...OptionFunc) factory.AppServerFactory

NewWorker create new postgres event listener

Types

type EventPayload

type EventPayload struct {
	EventID string           `json:"event_id"`
	Table   string           `json:"table"`
	Action  string           `json:"action"`
	Data    EventPayloadData `json:"data"`
}

EventPayload event model

func (EventPayload) GetID

func (e EventPayload) GetID() string

GetID get id if old/new data is empty, cause from long payload limitation

type EventPayloadData

type EventPayloadData struct {
	IsTooLongPayload bool        `json:"is_too_long_payload,omitempty"`
	OldID            string      `json:"old_id"`
	NewID            string      `json:"new_id"`
	Old              interface{} `json:"old"`
	New              interface{} `json:"new"`
}

EventPayloadData event data

type OptionFunc

type OptionFunc func(*option)

OptionFunc type

func SetDebugMode

func SetDebugMode(debugMode bool) OptionFunc

SetDebugMode option func

func SetLocker

func SetLocker(locker candiutils.Locker) OptionFunc

SetLocker option func

func SetMaxGoroutines

func SetMaxGoroutines(maxGoroutines int) OptionFunc

SetMaxGoroutines option func

func SetPostgresDSN

func SetPostgresDSN(dsn string) OptionFunc

SetPostgresDSN option func