xd_rsync

README

xd-rsync

xd-rsync is a tool to perform remote synchronisation of the XD ERP database with AWS SNS queues. The primary purpose of this tool is to enable businesses who use XD as their ERP to create event-driven architectures, by pushing updates of certain entities to specific SNS topics.

The SNS topics could then feed into many different SQS queues in order for different systems to process the messages at their discretion.

Usage

Prerequisites

1. Have an instance of XD ERP running alongside its local database.
2. Create config.json in the same folder where xd-rsync is going to be executed.

Config example

Customise your configuration file (config.json) according to your needs.

Datadog ingest URLs can be found on their documentation.

{
  "environment": "development",
  "awsRegion": "eu-west-2",
  // Database DSN for the xd-rsync client to be able to connect
  "dsn": "root:root@tcp(localhost:3306)/xd?charset=utf8mb4&parseTime=True&loc=Local",
  "queues": {
    // SNS topic for product updates to be published
    "productUpdatesSnsQueueArn": ""
  },
  "datadog": {
    // Datadog custom host
    "ingestHost": "http-intake.logs.datadoghq.eu",
    // Datadog API key
    "apiKey": "<INSERT_DATADOG_KEY_HERE>",

    // Insert fields that all events should contain
    "eventBaseFields": {
      "customEventProperty": "value"
    }
  },
  "syncFrequency": "5m" // Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".
}

Development

Setup

# Install all dependencies
go mod download

# Move to application entrypoint
cd cmd/xd-rsync

# Build & test xd-rsync
go build && ./xd-rsync

Local replica

The first step is to perform a database dump from XD database. After moving it into the folder dumps/, run:

# Start Docker-contained MySQL service
docker compose up

# Stop database container
docker compose down

# Reset database container by removing named volumes
docker compose down -v && docker compose up --force-recreate

# Run dump SQL script against container
mysql --max_allowed_packet=256M -h localhost -u root --protocol=tcp --password=root xd < ./dumps/dumpname.sql

Why use xd-rsync?

Despite supporting different databases engines, XD's Development & CS teams do not encourage their clients to host their databases on the cloud but rather OnPrem (on-premises).

Up to this point the only reason they can provide for being so strongly against cloud hosting, is that it's easier for them to maintain. Guess what, it's 2024, the PHP era has become the triassic era for devs, and distributed systems are more common than ever.

After all, not only is it cheaper and easier to maintain distributed systems (if you're doing it right) but also scales much better than hosting everything in the same place.

How does it work

The product changes are tracked using a combination of fields:

Change	DB Field
Product details change	Item sync timestamp: items.SyncStamp
Stock entry details change (not updated on entrance/exit)	Item stock sync timestamp: itemstock.SyncStamp
Stock entrance movement	Item stock last entrance movement timestamp: itemstock.LastEntrance
Stock exit movement	Item stock last exit movement timestamp: itemstock.LastExit

Filtering the products whose tracking fields have changed since the last update, xd-rsync is able to then push the updates to the SNS topic provided.

Documentation

Index

• Variables
• type Config
• type DatabaseService
• type DatadogConfig
• type MessagePublishInput
• type QueuesConfig
• type SNSService
• type XdProduct
  • func (p *XdProduct) GetKnownColumns() []string
  • func (p *XdProduct) GetKnownColumnsQuerySelectors() string
  • func (p *XdProduct) GetPrimaryKeyColumnName() string
  • func (p *XdProduct) GetTableName() string
  • func (p *XdProduct) ToJSON() (string, error)
• type XdProducts
  • func (ps *XdProducts) GetKnownColumnsQuerySelectors() string
  • func (ps *XdProducts) GetPrimaryKeyColumnName() string
  • func (ps *XdProducts) GetTableName() string
  • func (ps *XdProducts) ToJSON() (string, error)
• type XdProductsChunksWithMutex
  • func (c *XdProductsChunksWithMutex) GetList(list *XdProducts)
  • func (c *XdProductsChunksWithMutex) UpdateChunk(index int, productsList *XdProducts)
• type XdRsyncInstance
• type XdRsyncServices

Variables

var ErrProductJsonNotValid = fmt.Errorf("emitted product JSON is not valid")

Types

type Config

type Config struct {
	Environment      string         `json:"environment"`
	IsProductionMode bool           `json:"isProductionMode"`
	AwsRegion        string         `json:"awsRegion"`
	DSN              string         `json:"dsn"`
	Queues           *QueuesConfig  `json:"queues"`
	SyncFrequency    time.Duration  `json:"syncFrequency"`
	DatadogConfig    *DatadogConfig `json:"datadog"`
}

type DatabaseService

type DatabaseService interface {
	GetProductByReferece(id string) (*XdProduct, error)
	GetProductsByReferece(ids []string) (*XdProducts, error)
	GetPricedProductsCount(ts *time.Time) (int, error)
	GetPaginatedPricedProducts(ts *time.Time, limit int, offset int) (*XdProducts, error)
	GetPricedProducts() (*XdProducts, error)
	GetPricedProductsSinceTimestamp(ts *time.Time) (*XdProducts, error)
}

type DatadogConfig

type DatadogConfig struct {
	IngestHost      *string                 `json:"ingestHost"`
	ApiKey          *string                 `json:"datadogApiKey"`
	EventBaseFields *map[string]interface{} `json:"eventBaseFields"`
}

type MessagePublishInput

type MessagePublishInput struct {
	Message        string
	MessageGroupId string
}

type QueuesConfig

type QueuesConfig struct {
	ProductUpdatesSnsQueueArn string `json:"productUpdatesSnsQueueArn,omitempty"`
}

type SNSService

type SNSService interface {
	SendMessage(topicArn string, input *MessagePublishInput) error
	SendMessagesBatch(topicArn string, input *[]MessagePublishInput) (int, []error)
}

type XdProduct

type XdProduct struct {
	SKU               string     `db:"KeyId" dbSelector:"i.KeyId" json:"sku"`
	Description       string     `db:"Description" dbSelector:"i.Description" json:"name"`
	RetailPrice1      float64    `db:"RetailPrice1" dbSelector:"i.RetailPrice1" json:"clientCompareAtPrice"`
	RetailPrice2      float64    `db:"RetailPrice2" dbSelector:"i.RetailPrice2" json:"clientPrice"`
	AvailableQuantity float64    `db:"AvailableQuantity" dbSelector:"IFNULL(istock.AvailableQuantity, 0) as AvailableQuantity" json:"availableQuantity"`
	SyncStamp         *time.Time `db:"SyncStamp" dbSelector:"i.SyncStamp as SyncStamp" json:"syncStamp"`
	StockSyncStamp    *time.Time `db:"StockSyncStamp" dbSelector:"istock.SyncStamp as StockSyncStamp" json:"stockSyncStamp"`
	StockLastEntrance *time.Time `db:"StockLastEntrance" dbSelector:"istock.LastEntrance as StockLastEntrance" json:"stockLastEntrance"`
	StockLastExit     *time.Time `db:"StockLastExit" dbSelector:"istock.LastExit as StockLastExit" json:"stockLastExit"`
}

func (*XdProduct) GetKnownColumns

func (p *XdProduct) GetKnownColumns() []string

func (*XdProduct) GetKnownColumnsQuerySelectors

func (p *XdProduct) GetKnownColumnsQuerySelectors() string

func (*XdProduct) GetPrimaryKeyColumnName

func (p *XdProduct) GetPrimaryKeyColumnName() string

func (*XdProduct) GetTableName

func (p *XdProduct) GetTableName() string

func (*XdProduct) ToJSON

func (p *XdProduct) ToJSON() (string, error)

type XdProducts

type XdProducts []XdProduct

func (*XdProducts) GetKnownColumnsQuerySelectors

func (ps *XdProducts) GetKnownColumnsQuerySelectors() string

func (*XdProducts) GetPrimaryKeyColumnName

func (ps *XdProducts) GetPrimaryKeyColumnName() string

func (*XdProducts) GetTableName

func (ps *XdProducts) GetTableName() string

func (*XdProducts) ToJSON

func (ps *XdProducts) ToJSON() (string, error)

type XdProductsChunksWithMutex

type XdProductsChunksWithMutex struct {
	Chunks *map[int]XdProducts
	// contains filtered or unexported fields
}

func (*XdProductsChunksWithMutex) GetList

func (c *XdProductsChunksWithMutex) GetList(list *XdProducts)

func (*XdProductsChunksWithMutex) UpdateChunk

func (c *XdProductsChunksWithMutex) UpdateChunk(index int, productsList *XdProducts)

type XdRsyncInstance

type XdRsyncInstance struct {
	Config   *Config
	Logger   *logger.Logger
	Services *XdRsyncServices
}

type XdRsyncServices

type XdRsyncServices struct {
	Database DatabaseService
	SNS      SNSService
}