“

Series: S12 Developer Tutorials · Article 13 | Level: Intermediate | Reading Time: 15 min

Event Subscriptions & Notifications: Let the Platform Tell You

#Introduction

In previous tutorials, we learned how to actively query data, execute Actions, and trigger rules. But in real-world business scenarios, "waiting for notifications" is often far more efficient than "polling for changes" — automatically notifying downstream systems when an order status changes, instantly pushing alerts when a metric becomes anomalous, or triggering follow-up workflows when an approval completes.

The coomia-dip event subscription system is designed precisely for this. Built on object change events within the ontology model, it provides a declarative subscription mechanism. You simply tell the platform "what you care about," and the platform proactively pushes messages to you when events occur. This tutorial will guide you through building a complete event subscription and notification system from scratch.

#1. Understanding the coomia-dip Event Model

#1.1 The Nature of Events

In coomia-dip, every change is an event. When you modify an object's properties through an Action, create a new link relationship, or delete an object, the platform internally produces a corresponding change event. These events form a continuous Event Stream, and the subscription system's job is to filter out the events you care about and deliver them to you.

Python

from ontology_sdk.models import OntologyEvent

# Basic event structure
class OntologyEvent:
    """Ontology change event"""
    event_id: str               # Globally unique event ID
    event_type: EventType       # CREATE / UPDATE / DELETE / LINK / UNLINK
    object_type: str            # Object type that changed
    object_rid: str             # RID of the changed object
    timestamp: datetime         # Event timestamp (UTC)
    changes: dict[str, Change]  # Field-level change details
    actor: str                  # User or system that triggered the change
    transaction_id: str         # Parent transaction ID

#1.2 Event Types Explained

coomia-dip categorizes events into five basic types:

Event Type	Meaning	Trigger Scenario
`CREATE`	New object created	Object creation via Action, data import
`UPDATE`	Object property changed	Modification of one or more properties
`DELETE`	Object deleted	Soft delete or hard delete
`LINK`	Relationship established	Link created between two objects
`UNLINK`	Relationship removed	Link removed between two objects

Each event type carries complete change context, including the value before (before) and after (after) the change, allowing subscribers to precisely understand what happened.

#1.3 Event Delivery Guarantees

The coomia-dip event system provides the following guarantees:

At-Least-Once Delivery: Every event will be delivered at least once to every active subscription
Event Ordering: Events for the same object are guaranteed to be delivered in the order they occurred
Persistent Storage: Events are persisted to the event store before delivery; they survive system restarts
Replay Capability: Supports replaying events from any historical point in time

Python

# Event store persistence mechanism
class EventStore:
    """
    Event storage uses Apache Kafka as the underlying implementation:
    - Each ObjectType maps to a Kafka Topic
    - Partition key is object_rid, ensuring same-object event ordering
    - Default retention policy is 7 days, configurable as needed
    """

    def append(self, event: OntologyEvent) -> None:
        topic = f"onto.events.{event.object_type}"
        partition_key = event.object_rid
        self.kafka_producer.send(topic, key=partition_key, value=event.serialize())

    def replay(self, object_type: str, from_offset: int) -> Iterator[OntologyEvent]:
        """Replay events from a specified offset"""
        consumer = self.kafka_consumer.subscribe(f"onto.events.{object_type}")
        consumer.seek(from_offset)
        for record in consumer:
            yield OntologyEvent.deserialize(record.value)

#2. Creating Your First Subscription

#2.1 Creating a Subscription via SDK

Let's start with a practical scenario: you need to be notified when a "Ticket" object's status changes to "COMPLETED."

Python

from ontology_sdk import OntoPlatform
from ontology_sdk.subscription import (
    Subscription,
    EventFilter,
    FilterCondition,
    WebhookTarget,
)

# Initialize platform connection
platform = OntoPlatform(
    base_url="http://localhost:8080",
    token="your-api-token",
)

# Define subscription
subscription = Subscription(
    name="ticket-completed-notify",
    description="Notify downstream systems when tickets are completed",
    object_type="Ticket",
    event_types=["UPDATE"],
    filter=EventFilter(
        conditions=[
            FilterCondition(
                field="status",
                operator="eq",
                value="COMPLETED",
                apply_to="after",   # Filter on the post-change value
            ),
            FilterCondition(
                field="status",
                operator="neq",
                value="COMPLETED",
                apply_to="before",  # Was not COMPLETED before (prevent duplicate triggers)
            ),
        ],
        logic="AND",
    ),
    target=WebhookTarget(
        url="https://your-service.example.com/webhook/ticket-completed",
        method="POST",
        headers={"Authorization": "Bearer your-webhook-secret"},
        retry_policy={"max_retries": 3, "backoff_ms": 1000},
    ),
)

# Register subscription
result = platform.subscriptions.create(subscription)
print(f"Subscription created: {result.subscription_id}")
print(f"Status: {result.status}")

#2.2 Subscription Filter Conditions

Filter conditions are the core capability of the subscription system. coomia-dip supports rich filter expressions:

Python

# Example 1: Multi-condition composite filter
filter_complex = EventFilter(
    conditions=[
        FilterCondition(field="priority", operator="in", value=["HIGH", "CRITICAL"]),
        FilterCondition(field="assignee", operator="is_not_null"),
        FilterCondition(field="updated_at", operator="gt", value="2025-01-01T00:00:00Z"),
    ],
    logic="AND",
)

# Example 2: Nested filtering (OR + AND)
filter_nested = EventFilter(
    logic="OR",
    groups=[
        EventFilter(
            logic="AND",
            conditions=[
                FilterCondition(field="status", operator="eq", value="CRITICAL", apply_to="after"),
                FilterCondition(field="region", operator="eq", value="APAC"),
            ],
        ),
        EventFilter(
            logic="AND",
            conditions=[
                FilterCondition(field="status", operator="eq", value="DOWN", apply_to="after"),
                FilterCondition(field="sla_tier", operator="eq", value="PLATINUM"),
            ],
        ),
    ],
)

# Example 3: Delta magnitude filter
filter_delta = EventFilter(
    conditions=[
        FilterCondition(
            field="temperature",
            operator="delta_gt",     # Change magnitude greater than
            value=5.0,
            apply_to="delta",        # Compare |after - before|
        ),
    ],
)

#2.3 Delivery Target Types

coomia-dip supports multiple delivery targets:

Python

from ontology_sdk.subscription import (
    WebhookTarget,
    GrpcTarget,
    KafkaTarget,
    InternalActionTarget,
)

# 1. Webhook — most universal approach
webhook = WebhookTarget(
    url="https://api.example.com/events",
    method="POST",
    headers={"Content-Type": "application/json"},
    timeout_ms=5000,
    retry_policy={"max_retries": 3, "backoff_ms": 1000},
)

# 2. gRPC — high-performance scenarios
grpc_target = GrpcTarget(
    endpoint="event-handler.internal:50051",
    service="EventHandlerService",
    method="HandleOntologyEvent",
    tls_enabled=True,
)

# 3. Kafka — large-scale async processing
kafka_target = KafkaTarget(
    bootstrap_servers="kafka:9092",
    topic="downstream.events",
    key_expression="${event.object_rid}",
    partition_strategy="BY_KEY",
)

# 4. Internal Action — chain triggers
action_target = InternalActionTarget(
    action_type="NotifyStakeholders",
    parameter_mapping={
        "ticket_id": "${event.object_rid}",
        "new_status": "${event.changes.status.after}",
        "changed_by": "${event.actor}",
    },
)

#3. gRPC Streaming Subscriptions

#3.1 Server-Side Streaming

For scenarios requiring low latency and continuous event reception, coomia-dip provides a gRPC streaming subscription interface:

PROTOBUF

// subscription.proto
syntax = "proto3";

package onto.subscription.v1;

service SubscriptionService {
    // Create subscription
    rpc CreateSubscription(CreateSubscriptionRequest) returns (CreateSubscriptionResponse);

    // Stream events
    rpc StreamEvents(StreamEventsRequest) returns (stream OntologyEventMessage);

    // Acknowledge processed events
    rpc AcknowledgeEvents(AcknowledgeRequest) returns (AcknowledgeResponse);

    // Manage subscriptions
    rpc ListSubscriptions(ListRequest) returns (ListResponse);
    rpc PauseSubscription(PauseRequest) returns (PauseResponse);
    rpc ResumeSubscription(ResumeRequest) returns (ResumeResponse);
    rpc DeleteSubscription(DeleteRequest) returns (DeleteResponse);
}

message StreamEventsRequest {
    string subscription_id = 1;
    int64 from_offset = 2;        // Optional, start from specified position
    int32 batch_size = 3;          // Events per batch
    int32 heartbeat_interval_ms = 4;  // Heartbeat interval
}

message OntologyEventMessage {
    string event_id = 1;
    string event_type = 2;
    string object_type = 3;
    string object_rid = 4;
    int64 timestamp_ms = 5;
    map<string, FieldChange> changes = 6;
    string actor = 7;
    string transaction_id = 8;
    int64 offset = 9;             // Used for ACK
}

#3.2 Python Client Implementation

Python

import grpc
import asyncio
from ontology_sdk.grpc_client import SubscriptionServiceStub

async def stream_events():
    """Stream ticket change events"""

    channel = grpc.aio.insecure_channel("localhost:50051")
    stub = SubscriptionServiceStub(channel)

    request = StreamEventsRequest(
        subscription_id="sub-ticket-completed",
        batch_size=10,
        heartbeat_interval_ms=30000,
    )

    ack_batch = []

    async for event in stub.StreamEvents(request):
        # Heartbeat message (empty event)
        if not event.event_id:
            print("Heartbeat received")
            continue

        # Process event
        print(f"[{event.event_type}] {event.object_type}/{event.object_rid}")
        print(f"  Changes: {dict(event.changes)}")
        print(f"  Actor: {event.actor}")

        # Batch acknowledgment
        ack_batch.append(event.offset)
        if len(ack_batch) >= 10:
            await stub.AcknowledgeEvents(AcknowledgeRequest(
                subscription_id="sub-ticket-completed",
                offsets=ack_batch,
            ))
            ack_batch.clear()
            print("  Acknowledged batch")

    await channel.close()

# Run
asyncio.run(stream_events())

#3.3 Resilient Client with Reconnection

In production, network disconnections and service restarts are routine. Here is a robust implementation with automatic reconnection:

Python

import grpc
import asyncio
import logging
from typing import Callable, Awaitable

logger = logging.getLogger(__name__)

class ResilientEventSubscriber:
    """Event subscription client with automatic reconnection"""

    def __init__(
        self,
        endpoint: str,
        subscription_id: str,
        handler: Callable[[OntologyEventMessage], Awaitable[None]],
        max_retries: int = -1,          # -1 means infinite retries
        base_backoff_s: float = 1.0,
        max_backoff_s: float = 60.0,
    ):
        self.endpoint = endpoint
        self.subscription_id = subscription_id
        self.handler = handler
        self.max_retries = max_retries
        self.base_backoff_s = base_backoff_s
        self.max_backoff_s = max_backoff_s
        self._last_offset: int = 0
        self._running = False

    async def start(self):
        """Start subscription with automatic reconnection"""
        self._running = True
        retries = 0

        while self._running:
            try:
                channel = grpc.aio.insecure_channel(self.endpoint)
                stub = SubscriptionServiceStub(channel)

                request = StreamEventsRequest(
                    subscription_id=self.subscription_id,
                    from_offset=self._last_offset,
                    batch_size=20,
                    heartbeat_interval_ms=30000,
                )

                logger.info(f"Connecting to event stream from offset {self._last_offset}")

                async for event in stub.StreamEvents(request):
                    retries = 0  # Reset retry count on successful message

                    if not event.event_id:
                        continue  # Heartbeat

                    await self.handler(event)
                    self._last_offset = event.offset + 1

                    # Per-event acknowledgment (can be optimized to batch)
                    await stub.AcknowledgeEvents(AcknowledgeRequest(
                        subscription_id=self.subscription_id,
                        offsets=[event.offset],
                    ))

            except grpc.aio.AioRpcError as e:
                if e.code() == grpc.StatusCode.UNAVAILABLE:
                    logger.warning(f"Server unavailable, will retry: {e.details()}")
                elif e.code() == grpc.StatusCode.NOT_FOUND:
                    logger.error(f"Subscription not found: {self.subscription_id}")
                    break
                else:
                    logger.error(f"gRPC error: {e.code()} - {e.details()}")

            except Exception as e:
                logger.error(f"Unexpected error: {e}")

            finally:
                try:
                    await channel.close()
                except Exception:
                    pass

            if not self._running:
                break

            retries += 1
            if self.max_retries >= 0 and retries > self.max_retries:
                logger.error("Max retries exceeded, stopping subscriber")
                break

            backoff = min(self.base_backoff_s * (2 ** (retries - 1)), self.max_backoff_s)
            logger.info(f"Reconnecting in {backoff:.1f}s (attempt {retries})")
            await asyncio.sleep(backoff)

    async def stop(self):
        """Graceful shutdown"""
        self._running = False

# Usage example
async def handle_event(event: OntologyEventMessage):
    print(f"Processing: {event.event_type} on {event.object_rid}")
    # Your business logic here...

subscriber = ResilientEventSubscriber(
    endpoint="localhost:50051",
    subscription_id="sub-ticket-completed",
    handler=handle_event,
)

# Start
await subscriber.start()

#4. Real-World Scenario: Multi-System Integration

#4.1 Scenario Description

Imagine you are building an IT Service Management (ITSM) system and need to implement the following integrations:

Ticket created -> Notify Slack channel
Ticket assigned -> Send email to assignee
Ticket completed -> Trigger satisfaction survey
SLA approaching expiry -> Escalation alert

#4.2 Implementation

Python

from ontology_sdk import OntoPlatform
from ontology_sdk.subscription import Subscription, EventFilter, FilterCondition

platform = OntoPlatform(base_url="http://localhost:8080", token="your-token")

# Subscription 1: Ticket creation notification
sub_created = Subscription(
    name="ticket-created-slack",
    object_type="Ticket",
    event_types=["CREATE"],
    target=WebhookTarget(
        url="https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
        method="POST",
        body_template="""{
            "channel": "#itsm-tickets",
            "text": "New ticket <${event.object_rid}|${event.changes.title.after}> created\\nPriority: ${event.changes.priority.after}\\nCreated by: ${event.actor}"
        }""",
    ),
)

# Subscription 2: Ticket assignment notification
sub_assigned = Subscription(
    name="ticket-assigned-email",
    object_type="Ticket",
    event_types=["UPDATE"],
    filter=EventFilter(
        conditions=[
            FilterCondition(field="assignee", operator="is_not_null", apply_to="after"),
            FilterCondition(field="assignee", operator="changed"),
        ],
        logic="AND",
    ),
    target=WebhookTarget(
        url="https://email-service.internal/send",
        method="POST",
        body_template="""{
            "to": "${event.changes.assignee.after}",
            "subject": "New ticket assigned: ${event.changes.title.after}",
            "body": "You have been assigned ticket ${event.object_rid}, priority ${event.changes.priority.after}"
        }""",
    ),
)

# Subscription 3: Ticket completion triggers survey
sub_completed = Subscription(
    name="ticket-completed-survey",
    object_type="Ticket",
    event_types=["UPDATE"],
    filter=EventFilter(
        conditions=[
            FilterCondition(field="status", operator="eq", value="COMPLETED", apply_to="after"),
            FilterCondition(field="status", operator="neq", value="COMPLETED", apply_to="before"),
        ],
        logic="AND",
    ),
    target=InternalActionTarget(
        action_type="SendSatisfactionSurvey",
        parameter_mapping={
            "ticket_id": "${event.object_rid}",
            "reporter": "${event.changes.reporter.after}",
            "resolved_by": "${event.changes.assignee.after}",
        },
    ),
)

# Subscription 4: SLA warning
sub_sla_warning = Subscription(
    name="ticket-sla-warning",
    object_type="Ticket",
    event_types=["UPDATE"],
    filter=EventFilter(
        conditions=[
            FilterCondition(field="sla_remaining_hours", operator="lt", value=2),
            FilterCondition(field="status", operator="in", value=["OPEN", "IN_PROGRESS"]),
        ],
        logic="AND",
    ),
    target=WebhookTarget(
        url="https://pagerduty.example.com/v2/enqueue",
        method="POST",
        body_template="""{
            "routing_key": "YOUR_PD_KEY",
            "event_action": "trigger",
            "payload": {
                "summary": "SLA Warning: Ticket ${event.object_rid} has <2h remaining",
                "severity": "warning",
                "source": "coomia-dip"
            }
        }""",
    ),
)

# Batch registration
for sub in [sub_created, sub_assigned, sub_completed, sub_sla_warning]:
    result = platform.subscriptions.create(sub)
    print(f"Created: {result.subscription_id} -> {result.status}")

#5. Subscription Management & Monitoring

#5.1 Viewing Subscription Status

Python

# List all subscriptions
subscriptions = platform.subscriptions.list()
for sub in subscriptions:
    print(f"{sub.name}: {sub.status}")
    print(f"  Created: {sub.created_at}")
    print(f"  Events delivered: {sub.stats.total_delivered}")
    print(f"  Events failed: {sub.stats.total_failed}")
    print(f"  Last event at: {sub.stats.last_event_at}")
    print(f"  Lag: {sub.stats.current_lag} events")
    print()

# Detailed statistics for a specific subscription
stats = platform.subscriptions.get_stats("sub-ticket-completed")
print(f"Delivery success rate: {stats.success_rate:.1%}")
print(f"Average latency: {stats.avg_latency_ms:.0f}ms")
print(f"P99 latency: {stats.p99_latency_ms:.0f}ms")

#5.2 Pause and Resume

Python

# Pause subscription (useful during maintenance)
platform.subscriptions.pause("sub-ticket-completed")
print("Subscription paused")

# Resume subscription (continues from pause point)
platform.subscriptions.resume("sub-ticket-completed")
print("Subscription resumed, catching up...")

# Reset subscription offset (reprocess historical events)
platform.subscriptions.reset_offset(
    "sub-ticket-completed",
    to_timestamp="2025-01-01T00:00:00Z",
)
print("Offset reset, replaying from 2025-01-01")

#5.3 Dead Letter Queue (DLQ)

Events that fail delivery beyond the retry limit are not lost — they enter the dead letter queue:

Python

# View events in the dead letter queue
dead_letters = platform.subscriptions.list_dead_letters("sub-ticket-completed")
for dl in dead_letters:
    print(f"Event: {dl.event_id}")
    print(f"  Failed at: {dl.failed_at}")
    print(f"  Attempts: {dl.attempt_count}")
    print(f"  Last error: {dl.last_error}")
    print()

# Retry dead letter events
platform.subscriptions.retry_dead_letters(
    "sub-ticket-completed",
    event_ids=[dl.event_id for dl in dead_letters[:10]],
)
print("Retrying 10 dead letter events")

# Purge expired dead letters
platform.subscriptions.purge_dead_letters(
    "sub-ticket-completed",
    before="2025-01-01T00:00:00Z",
)

#6. Advanced Patterns

#6.1 Aggregated Event Subscriptions

Sometimes you don't want every individual change event — instead, you'd like a summary delivered periodically:

Python

from ontology_sdk.subscription import AggregationWindow

sub_daily_summary = Subscription(
    name="ticket-daily-summary",
    object_type="Ticket",
    event_types=["CREATE", "UPDATE"],
    aggregation=AggregationWindow(
        window_size="1h",             # Aggregate every hour
        aggregate_fields={
            "total_created": {"type": "count", "filter": {"event_type": "CREATE"}},
            "total_resolved": {
                "type": "count",
                "filter": {
                    "event_type": "UPDATE",
                    "changes.status.after": "COMPLETED",
                },
            },
            "avg_resolution_time_hours": {
                "type": "avg",
                "field": "resolution_time_hours",
                "filter": {"event_type": "UPDATE", "changes.status.after": "COMPLETED"},
            },
        },
    ),
    target=WebhookTarget(
        url="https://dashboard.example.com/api/metrics",
        method="POST",
    ),
)

#6.2 Cross-Object Correlated Subscriptions

Subscriptions are not limited to a single object type. You can set up cross-object integrations through link relationships:

Python

# When any Ticket under a Project completes, check if ALL tickets are complete
sub_project_completion = Subscription(
    name="project-all-tickets-completed",
    object_type="Ticket",
    event_types=["UPDATE"],
    filter=EventFilter(
        conditions=[
            FilterCondition(field="status", operator="eq", value="COMPLETED", apply_to="after"),
        ],
    ),
    correlation=CorrelationCheck(
        traverse_link="belongsToProject",
        target_object_type="Project",
        condition="ALL_LINKED_MATCH",
        linked_type="Ticket",
        linked_filter=FilterCondition(field="status", operator="eq", value="COMPLETED"),
    ),
    target=InternalActionTarget(
        action_type="MarkProjectCompleted",
        parameter_mapping={
            "project_rid": "${correlation.target_rid}",
        },
    ),
)

#6.3 Conditional Deduplication

Prevent duplicate triggers within a short time window:

Python

sub_with_dedup = Subscription(
    name="alert-dedup",
    object_type="MonitoringAlert",
    event_types=["CREATE"],
    deduplication=DeduplicationConfig(
        key_expression="${event.changes.alert_type.after}:${event.changes.host.after}",
        window_seconds=300,  # Same key triggers only once within 5 minutes
        strategy="FIRST",    # Keep the first event, or LAST for the latest
    ),
    target=WebhookTarget(url="https://alert-handler.internal/handle"),
)

#7. Debugging & Troubleshooting

#7.1 Event Replay Testing

During development, you can use event replay to validate subscription logic:

Python

# Simulate event delivery (dry-run mode)
test_event = OntologyEvent(
    event_type="UPDATE",
    object_type="Ticket",
    object_rid="ri.ticket.123",
    changes={
        "status": Change(before="IN_PROGRESS", after="COMPLETED"),
        "resolution_time_hours": Change(before=None, after=4.5),
    },
    actor="test-user",
)

result = platform.subscriptions.test_delivery(
    subscription_id="sub-ticket-completed",
    event=test_event,
    dry_run=True,  # Don't actually deliver, just return match result
)

print(f"Would match: {result.matched}")
print(f"Would deliver to: {result.target_description}")
print(f"Payload preview: {result.payload_preview}")

#7.2 Viewing Delivery Logs

Python

# Get recent delivery logs
logs = platform.subscriptions.get_delivery_logs(
    "sub-ticket-completed",
    limit=20,
    status="FAILED",  # Only show failures
)

for log in logs:
    print(f"Event: {log.event_id}")
    print(f"  Time: {log.timestamp}")
    print(f"  Status: {log.status}")
    print(f"  Response code: {log.response_code}")
    print(f"  Response body: {log.response_body[:200]}")
    print(f"  Duration: {log.duration_ms}ms")
    print()

#7.3 Common Issues

Problem	Cause	Solution
Subscription not triggering	Filter conditions too strict	Use `test_delivery` to validate filter logic
Duplicate triggers	No deduplication configured	Add `deduplication` configuration
High latency	Slow consumer-side processing	Check webhook response time, increase concurrency
Missing events	Not acknowledged (ACK)	Ensure gRPC streaming client properly ACKs
Out-of-order events	Cross-partition consumption	Ensure same-object events land in same partition

#8. Production Best Practices

#8.1 Subscription Design Principles

Single Responsibility: Each subscription does one thing; avoid "catch-all" subscriptions
Idempotent Consumption: Consumer side must support idempotency since events may be delivered more than once
Fast Return: Webhook handlers should return within 5 seconds; offload long tasks asynchronously
Monitor Lag: Watch subscription consumer lag; recommended alert threshold is 1000 events
Graceful Degradation: Consumer unavailability should not affect event production

#8.2 Performance Tuning

Python

# Subscription configuration for high-throughput scenarios
high_throughput_sub = Subscription(
    name="high-volume-processor",
    object_type="SensorReading",
    event_types=["CREATE"],
    performance=PerformanceConfig(
        parallelism=8,              # Parallel consumer threads
        batch_size=100,             # Batch delivery size
        batch_timeout_ms=1000,      # Batch timeout (deliver on size or timeout)
        max_in_flight=1000,         # Max in-flight messages
    ),
    target=KafkaTarget(
        bootstrap_servers="kafka:9092",
        topic="sensor.events.processed",
        compression="lz4",
    ),
)

#8.3 Security Recommendations

Signature Verification: Webhook deliveries carry HMAC signatures; consumers should verify them
TLS Encryption: All webhook and gRPC connections must use TLS in production
Least Privilege: Subscription tokens should only grant read access to specific ObjectType events
Audit Logging: All subscription creation, modification, and deletion operations are recorded in the audit log

Python

# Webhook signature verification example (consumer side)
import hmac
import hashlib

def verify_webhook(request, secret: str) -> bool:
    signature = request.headers.get("X-Onto-Signature")
    if not signature:
        return False

    expected = hmac.new(
        secret.encode(),
        request.body,
        hashlib.sha256,
    ).hexdigest()

    return hmac.compare_digest(f"sha256={expected}", signature)

#Summary

This tutorial covered the core features of the coomia-dip event subscription system:

Event Model: Understanding the five event types — CREATE/UPDATE/DELETE/LINK/UNLINK
Subscription Creation: Declaratively creating subscriptions via SDK with filter conditions and delivery targets
gRPC Streaming Subscriptions: Implementing low-latency real-time event consumption with reconnection
Real-World Scenario: Complete implementation of ITSM multi-system integration
Advanced Patterns: Event aggregation, cross-object correlation, conditional deduplication
Operations Management: Status monitoring, dead letter queues, pause and resume

Event subscriptions are the foundation for building reactive systems. In the next tutorial, we will explore how to extend the platform's capabilities through custom gRPC services.

Next: [S12-14] gRPC Custom Service Development Guide Previous: [S12-12] Data Import Best Practices