Event-driven sync

Keeping data in sync

Do not re-upload your catalog four times a day. A 200 000-article catalog changes by a few dozen records per day; sending all of them costs hours of traffic and re-indexes every customer record for nothing. Send changes when they happen.

Full dumps are the expensive default everyone starts with. Events are the primary path; reconciliation is the safety net.

The pattern

One change in the ERP → one request to AIOTIC.

1. Your ERP produces a change event (customer/product/mapping created, updated, deactivated).
2. Your integration service maps it to the AIOTIC shape and computes a fingerprint.
3. If the fingerprint differs from the last one sent, PUT (or DELETE) the record. Otherwise skip.
4. Remember the fingerprint.

The SDK's SyncEngine implements 2–4; you implement 1 with whatever your ERP offers.

Where change events come from

ERP capability	How to hook it	Latency
Outbound webhooks (Business Central, Exact Online, Odoo, Shopify-style)	Subscribe to customer.*, item.* events; POST them to your service's /erp/events	seconds
Outbox table / event log in the ERP database	A small worker reads new rows and calls the sync engine	seconds–minutes
Change Data Capture (SQL Server CDC, Debezium)	Consume the change stream	seconds
updated_at columns	Poll WHERE updated_at > :watermark every few minutes	minutes
Nothing (legacy)	Hash-based reconciliation on a schedule	hours

Receiving events in your service

The SDK ships a generic event endpoint (POST /erp/events) that accepts a small JSON format and feeds the engine — wire your ERP's webhook or an outbox worker to it:

POST /erp/events
X-API-KEY: <your own key for this endpoint>
{
  "events": [
    { "kind": "product",  "op": "upsert", "item_number": "620206_01", "language_code": "nl", "description": "Kabel 3x1.5 mm² (100 m)" },
    { "kind": "customer", "op": "upsert", "number": "58931", "name": "LUMITECH INSTALLATIES", "city": "Apeldoorn", "vat_number": "NL001234567B01" },
    { "kind": "mapping",  "op": "upsert", "customer_number": "58931", "customer_item_number": "LT-ART-001", "item_number": "PROD-001", "language_code": "nl" },
    { "kind": "customer", "op": "delete", "number": "10577" }
  ]
}

Or call the engine directly from your own code:

from aiotic.sync import ChangeEvent, SyncEngine, HashStateStore

engine = SyncEngine(client, state=HashStateStore("sync-state.db"))

def on_item_changed(item):                       # your ERP's callback
    if item.blocked or item.discontinued:
        engine.apply(ChangeEvent.product_delete(item.no, "nl"))
    else:
        engine.apply(ChangeEvent.product_upsert(item.no, "nl", description=item.description))

Polling on updated_at

For ERPs without events, the engine has a watermark-based poller:

from aiotic.sync import PollingChangeSource, ChangeEvent

def fetch_changed_since(watermark):
    rows = db.query("SELECT no, description, blocked, updated_at FROM items WHERE updated_at > ? ORDER BY updated_at", watermark or "1970-01-01")
    events = [ChangeEvent.product_delete(r.no, "nl") if r.blocked else ChangeEvent.product_upsert(r.no, "nl", description=r.description) for r in rows]
    return events, (rows[-1].updated_at.isoformat() if rows else watermark)

PollingChangeSource("items", fetch_changed_since, engine, interval=120).run_forever()

The watermark only advances when every event in the batch succeeded, so a transient failure is retried on the next run instead of being lost.

Mapping rules that keep AIOTIC accurate

ERP situation	AIOTIC action
Article blocked / discontinued	DELETE /product — it must stop being a valid article
Article re-activated	PUT /product
Customer deactivated	DELETE /customer (or rename with a legacy marker — see Archiving)
Customer merged into another	rename the old one *** ZIE <new number> *** so orders redirect, or delete it
Address / VAT / e-mail changed	PUT /customer — identification uses these fields
New language description	PUT /product/{no}/{lang}
Operator fixed an article in AIOTIC and you learned the customer's code	PUT /customer-product so the next order resolves automatically

Throttling and ordering

• The engine sends customers first, then products, then item mappings, because a mapping must reference an existing customer and product. Deletes go the other way round.
• Keep concurrency at 4–8. The client's token bucket (default 10 req/s) protects the tenant.
• Events for the same record are applied in the order you pass them and never concurrently, so a delete followed by a re-create ends with the record present. Across records the engine keeps the dependencies: a mapping after the customer and product events before it, a customer or product delete after the mapping events before it. Deliver full records (not diffs) in source order; the fingerprint is computed from the record, so the last one wins.