“

Series: S3 Data Foundation · Article 22 | Level: Advanced | Reading Time: 20 min

Subscription System: Real-Time Data Change Notifications

Tags: #Subscription #Realtime #ChangeNotification #WebSocket #EventDriven #coomia-dip

#TL;DR

The coomia-dip subscription system enables users and applications to subscribe to Ontology data change events — when specific Entity Type properties change, relationships are created or deleted, or metrics exceed thresholds, the system automatically pushes notifications. This article fully dissects the subscription system's architecture, including change event capture (via Iceberg Changelog and Flink CDC), subscription rule definition and matching engine, multi-channel push (WebSocket, SSE, Webhook, message queues), event deduplication and ordering guarantees, subscription performance optimization, and typical application scenarios.

#1. Subscription System Architecture

Code

Subscription System Architecture:

+--------------------------------------------------+
|                 Data Sources                      |
|  +----------+  +----------+  +----------+        |
|  | Flink CDC|  | Iceberg  |  | World    |        |
|  | Events   |  | Changelog|  | Transform|        |
|  +----------+  +----------+  +----------+        |
+-----------------------+--------------------------+
                        | Events
                        v
+--------------------------------------------------+
|              Event Processing Pipeline            |
|  +----------+  +----------+  +----------+        |
|  | Dedup    |->| Enrich   |->| Match    |        |
|  | & Order  |  | (Context)|  | (Rules)  |        |
|  +----------+  +----------+  +----------+        |
+-----------------------+--------------------------+
                        | Matched Events
                        v
+--------------------------------------------------+
|              Notification Dispatch                |
|  +----------+ +--------+ +--------+ +----+      |
|  |WebSocket | | SSE    | |Webhook | | MQ |      |
|  +----------+ +--------+ +--------+ +----+      |
+--------------------------------------------------+

#2. Subscription Rule Definition

#2.1 Subscription API

Python

from onto_subscription import Subscription, EventFilter, Channel

# Example 1: Monitor high-risk customer changes
sub1 = Subscription(
    name="high-risk-customer-watch",
    filters=[
        EventFilter(
            event_type="ENTITY_UPDATED",
            entity_type="Customer",
            property_filter={"risk_level": "HIGH"},
        )
    ],
    channels=[
        Channel.webhook("https://alert.company.com/risk"),
        Channel.email("risk-team@company.com"),
    ],
    debounce_seconds=30,
)

# Example 2: Device offline alert
sub2 = Subscription(
    name="device-offline-alert",
    filters=[
        EventFilter(
            event_type="ENTITY_UPDATED",
            entity_type="Device",
            property_changes={
                "status": {"from": "online", "to": "offline"}
            },
        )
    ],
    channels=[
        Channel.websocket(room="ops-dashboard"),
    ],
    debounce_seconds=0,
)

# Example 3: Metric threshold alert
sub3 = Subscription(
    name="cpu-threshold-alert",
    filters=[
        EventFilter(
            event_type="METRIC_THRESHOLD",
            entity_type="Server",
            metric_name="cpu_utilization",
            threshold={"operator": ">", "value": 90},
        )
    ],
    channels=[Channel.pagerduty(service_key="xxx")],
    cooldown_minutes=15,
)

#2.2 Rule Matching Engine

Python

class SubscriptionMatcher:
    """Subscription rule matching engine"""

    def __init__(self):
        self._subscriptions: list[Subscription] = []
        self._index: dict[str, list[Subscription]] = {}

    def register(self, subscription: Subscription):
        self._subscriptions.append(subscription)
        for f in subscription.filters:
            key = f"{f.event_type}:{f.entity_type}"
            self._index.setdefault(key, []).append(subscription)

    def match(self, event: OntologyEvent) -> list[MatchResult]:
        key = f"{event.type}:{event.entity_type}"
        candidates = self._index.get(key, [])

        matches = []
        for sub in candidates:
            for filter_rule in sub.filters:
                if self._matches_filter(event, filter_rule):
                    matches.append(MatchResult(
                        subscription=sub, event=event
                    ))
                    break

        return matches

    def _matches_filter(self, event, f) -> bool:
        if f.property_filter:
            for prop, expected in f.property_filter.items():
                if event.entity_properties.get(prop) != expected:
                    return False

        if f.property_changes:
            for prop, change in f.property_changes.items():
                if prop not in event.changed_properties:
                    return False
                if ('from' in change
                        and event.old_values.get(prop) != change['from']):
                    return False
                if ('to' in change
                        and event.new_values.get(prop) != change['to']):
                    return False

        if f.threshold:
            op = f.threshold['operator']
            val = f.threshold['value']
            if op == '>' and not (event.metric_value > val):
                return False
            if op == '<' and not (event.metric_value < val):
                return False

        return True

#3. Notification Push

#3.1 WebSocket Push

Python

class WebSocketNotifier:
    def __init__(self):
        self._connections: dict[str, set[WebSocket]] = {}

    async def notify(self, room: str, event: NotificationEvent):
        connections = self._connections.get(room, set())
        dead = set()
        for ws in connections:
            try:
                await ws.send_json(event.to_dict())
            except ConnectionClosed:
                dead.add(ws)
        for ws in dead:
            connections.discard(ws)

#3.2 Webhook Push (with retry)

Python

class WebhookNotifier:
    async def notify(self, url: str, event: NotificationEvent):
        payload = event.to_dict()
        for attempt in range(3):
            try:
                async with aiohttp.ClientSession() as session:
                    resp = await session.post(
                        url, json=payload,
                        headers={
                            'X-Onto-Event-Type': event.type,
                            'X-Onto-Signature': self._sign(payload),
                        },
                        timeout=aiohttp.ClientTimeout(total=10),
                    )
                    if resp.status < 300:
                        return
                    if resp.status >= 500:
                        await asyncio.sleep(2 ** attempt)
                        continue
                    return
            except Exception:
                await asyncio.sleep(2 ** attempt)
        await self._dead_letter_queue.put(event)

#4. Event Deduplication and Ordering

Python

class EventDeduplicator:
    def __init__(self, window_seconds: int = 60):
        self._seen: dict[str, datetime] = {}
        self._window = timedelta(seconds=window_seconds)

    def deduplicate(self, events: list[OntologyEvent]) -> list[OntologyEvent]:
        unique = []
        for event in sorted(events, key=lambda e: e.timestamp):
            key = f"{event.entity_id}:{event.type}:{event.property_hash}"
            last_seen = self._seen.get(key)
            if last_seen and event.timestamp - last_seen < self._window:
                continue
            self._seen[key] = event.timestamp
            unique.append(event)
        self._cleanup_expired()
        return unique

#5. Performance Metrics

Code

Subscription System Performance:

+-----------------+----------+----------+
| Metric           | Target   | Actual   |
+-----------------+----------+----------+
| Event-to-notify  | < 1s     | 200ms    |
| Concurrent subs  | 10,000   | 15,000   |
| Event throughput  | 10K/s    | 25K/s    |
| WebSocket conns   | 5,000    | 8,000    |
| Match latency     | < 1ms    | 0.3ms    |
+-----------------+----------+----------+

#6. Testing Strategy

Python

class TestSubscriptionSystem:

    async def test_property_change_triggers_notification(self):
        sub = Subscription(
            filters=[EventFilter(
                event_type="ENTITY_UPDATED",
                entity_type="Device",
                property_changes={"status": {"to": "offline"}}
            )],
            channels=[Channel.memory_queue("test")]
        )
        matcher.register(sub)

        event = OntologyEvent(
            type="ENTITY_UPDATED",
            entity_type="Device",
            entity_id="d1",
            changed_properties=["status"],
            old_values={"status": "online"},
            new_values={"status": "offline"},
        )
        matches = matcher.match(event)
        assert len(matches) == 1

    async def test_debounce_prevents_flooding(self):
        sub = Subscription(
            filters=[EventFilter(
                event_type="ENTITY_UPDATED",
                entity_type="Sensor"
            )],
            channels=[Channel.memory_queue("test")],
            debounce_seconds=5,
        )
        for i in range(100):
            await process_event(make_event(f"s{i}"))
        notifications = get_queue("test")
        assert len(notifications) <= 10

    async def test_deduplication(self):
        dedup = EventDeduplicator(window_seconds=10)
        events = [make_event("e1", ts=i) for i in range(5)]
        unique = dedup.deduplicate(events)
        assert len(unique) == 1

#Key Takeaways

The subscription system upgrades Ontology from "passive query" to "active push": users and apps don't need to poll for data changes; the system auto-notifies when data changes.
Flexible event filters support fine-grained subscriptions: subscribe by Entity Type, property values, change direction, metric thresholds, and more.
Multi-channel push covers different scenarios: WebSocket for real-time dashboards, Webhook for system integration, email/messaging for ops alerts.
Event deduplication and debouncing prevent notification floods: dedup windows and debounce mechanisms ensure high-frequency change scenarios don't create notification storms.
End-to-end event-to-notification latency < 1 second: from data change to notification push completes in sub-second, meeting real-time scenario requirements.

#Next Article

Next up: S3-23 "Data Export: Multi-Format Batch and Streaming Output" will show how to export data from Ontology to CSV, Parquet, Excel, and other formats.

Tags: #Subscription #Realtime #ChangeNotification #WebSocket #Webhook #EventDriven #Debounce #coomia-dip #DataFoundation