Batch processing (hourly or daily) works for reporting. Real-time processing enables:<\/p>\n\n\n\n The simplest pattern. Your webhook receiver processes each event inline.<\/p>\n\n\n\n Pros<\/strong>: Simple, no infrastructure beyond the receiver.\nCons<\/strong>: Processing time adds up. If Best for<\/strong>: Low-volume integrations (under 10 events\/second) with fast processing logic.<\/p>\n\n\n\n Decouple ingestion from processing using a message queue.<\/p>\n\n\n\n Each worker pulls from the queue (or a fan-out topic) and handles its own concern. Workers are independent: if the CRM integration is slow, it doesn't affect analytics ingestion.<\/p>\n\n\n\n Best for<\/strong>: Medium-volume integrations with multiple destinations or processing steps.<\/p>\n\n\n\n For high-volume, low-latency requirements, use a streaming platform like Kafka or Amazon Kinesis.<\/p>\n\n\n\n The stream processor can:<\/p>\n\n\n\n Best for<\/strong>: High-volume integrations (100+ events\/second) where you need windowed aggregations or stream joins.<\/p>\n\n\n\n Raw webhook events are useful but limited. Enrichment adds context that makes events actionable.<\/p>\n\n\n\n The MaxMind's GeoLite2<\/a> database is free and updates weekly. Tolinku uses the same database for its built-in analytics.<\/p>\n\n\n\n The webhook carries a For referral events, resolve the referral token to a user profile:<\/p>\n\n\n\n After enrichment, route events to different destinations based on type and content.<\/p>\n\n\n\n Each route runs independently. A Slack notification failure doesn't prevent the warehouse write.<\/p>\n\n\n\n Some insights require looking at events in aggregate, not individually. Windowed processing groups events by time and computes metrics over each window.<\/p>\n\n\n\n For production workloads, use a proper stream processing framework (Kafka Streams, Apache Flink, or AWS Kinesis Data Analytics) instead of in-memory windows. But for moderate volumes, the in-memory approach works and avoids infrastructure complexity.<\/p>\n\n\n\n Webhooks are delivered at-least-once. Tolinku retries failed deliveries (3 retries at 1 minute, 5 minutes, and 30 minutes). Your processing pipeline must handle duplicates.<\/p>\n\n\n\n For event storage (warehouse, database), deduplicate on insert using a hash of the event content:<\/p>\n\n\n\n For side effects (emails, Slack notifications, CRM updates), check the dedup table before acting:<\/p>\n\n\n\n See the webhook idempotency guide<\/a> for a deeper treatment.<\/p>\n\n\n\n![\"Tolinku](\"https:\/\/tolinku.com\/blog\/wp-content\/uploads\/2026\/03\/platform-webhooks.png\")
\nThe webhooks page with create form, webhook list, and delivery log.<\/em><\/p>\n\n\n\nWhy Real-Time?<\/h2>\n\n\n\n
\n
Architecture Patterns<\/h2>\n\n\n\n
Pattern 1: Direct Processing<\/h3>\n\n\n\n
Tolinku \u2192 Receiver \u2192 Process \u2192 Store\/Act\n<\/code><\/pre>\n\n\n\napp.post('\/webhooks\/tolinku', async (req, res) => {\n \/\/ Verify signature...\n res.status(200).send('OK');\n\n const event = JSON.parse(req.body.toString());\n\n \/\/ Process inline\n await enrichEvent(event);\n await routeToDestinations(event);\n await checkAlertConditions(event);\n});\n<\/code><\/pre>\n\n\n\nenrichEvent<\/code> takes 200ms and routeToDestinations<\/code> takes 300ms, you're spending 500ms per event. At 100 events\/second, you need concurrency or you'll fall behind.<\/p>\n\n\n\nPattern 2: Queue-Based Fan-Out<\/h3>\n\n\n\n
Tolinku \u2192 Receiver \u2192 Queue \u2192 Worker 1 (analytics)\n \u2192 Worker 2 (alerts)\n \u2192 Worker 3 (CRM)\n<\/code><\/pre>\n\n\n\nimport { SQSClient, SendMessageCommand } from '@aws-sdk\/client-sqs';\n\nconst sqs = new SQSClient({});\n\napp.post('\/webhooks\/tolinku', async (req, res) => {\n \/\/ Verify signature...\n res.status(200).send('OK');\n\n await sqs.send(new SendMessageCommand({\n QueueUrl: process.env.SQS_QUEUE_URL!,\n MessageBody: req.body.toString(),\n MessageAttributes: {\n EventType: {\n DataType: 'String',\n StringValue: req.headers['x-webhook-event'] as string,\n },\n },\n }));\n});\n<\/code><\/pre>\n\n\n\nPattern 3: Stream Processing<\/h3>\n\n\n\n
Tolinku \u2192 Receiver \u2192 Kafka Topic \u2192 Stream Processor \u2192 Destinations\n<\/code><\/pre>\n\n\n\n\n
Event Enrichment<\/h2>\n\n\n\n
Geo-Enrichment<\/h3>\n\n\n\n
link.clicked<\/code> event includes an IP address. Resolve it to a country, region, and city:<\/p>\n\n\n\nimport { Reader } from '@maxmind\/geoip2-node';\n\nlet geoReader: Reader;\n\nasync function initGeo() {\n geoReader = await Reader.open('\/path\/to\/GeoLite2-City.mmdb');\n}\n\nfunction enrichWithGeo(event: any): any {\n if (!event.data.ip) return event;\n\n try {\n const geo = geoReader.city(event.data.ip);\n return {\n ...event,\n data: {\n ...event.data,\n country: geo.country?.isoCode || null,\n region: geo.subdivisions?.[0]?.isoCode || null,\n city: geo.city?.names?.en || null,\n },\n };\n } catch {\n return event;\n }\n}\n<\/code><\/pre>\n\n\n\nCampaign Metadata<\/h3>\n\n\n\n
campaign<\/code> string. Your campaign management system has richer metadata: budget, target audience, creative variant, channel. Join on the campaign ID:<\/p>\n\n\n\nasync function enrichWithCampaign(event: any): Promise<any> {\n if (!event.data.campaign) return event;\n\n const campaign = await db.query(\n 'SELECT channel, budget, target_audience FROM campaigns WHERE campaign_id = $1',\n [event.data.campaign]\n );\n\n if (campaign.rows.length === 0) return event;\n\n return {\n ...event,\n data: {\n ...event.data,\n channel: campaign.rows[0].channel,\n budget: campaign.rows[0].budget,\n target_audience: campaign.rows[0].target_audience,\n },\n };\n}\n<\/code><\/pre>\n\n\n\nUser Identity<\/h3>\n\n\n\n
async function enrichWithUser(event: any): Promise<any> {\n if (!event.data.referrer_token) return event;\n\n const user = await db.query(\n 'SELECT id, email, name FROM users WHERE referral_token = $1',\n [event.data.referrer_token]\n );\n\n if (user.rows.length === 0) return event;\n\n return {\n ...event,\n data: {\n ...event.data,\n referrer_id: user.rows[0].id,\n referrer_name: user.rows[0].name,\n },\n };\n}\n<\/code><\/pre>\n\n\n\nEvent Routing<\/h2>\n\n\n\n
interface EventRouter {\n condition: (event: any) => boolean;\n handler: (event: any) => Promise<void>;\n}\n\nconst routes: EventRouter[] = [\n \/\/ All events go to the data warehouse\n {\n condition: () => true,\n handler: sendToWarehouse,\n },\n \/\/ Install events trigger a welcome workflow\n {\n condition: (e) => e.event === 'install.tracked',\n handler: triggerWelcomeWorkflow,\n },\n \/\/ Referral completions update the CRM\n {\n condition: (e) => e.event === 'referral.completed',\n handler: updateCRM,\n },\n \/\/ High-value events go to Slack\n {\n condition: (e) => ['install.tracked', 'referral.completed'].includes(e.event),\n handler: notifySlack,\n },\n \/\/ Anomalous click patterns trigger alerts\n {\n condition: (e) => e.event === 'link.clicked' && e.data.country === 'XX',\n handler: flagSuspiciousClick,\n },\n];\n\nasync function routeEvent(event: any) {\n const enriched = await enrichEvent(event);\n\n const promises = routes\n .filter(route => route.condition(enriched))\n .map(route => route.handler(enriched).catch(err =>\n console.error(`Route handler failed: ${err.message}`)\n ));\n\n await Promise.allSettled(promises);\n}\n<\/code><\/pre>\n\n\n\nWindowed Aggregations<\/h2>\n\n\n\n
Sliding Window: Clicks Per Campaign Per Minute<\/h3>\n\n\n\n
interface WindowBucket {\n campaign: string;\n count: number;\n firstSeen: number;\n}\n\nconst windows = new Map<string, WindowBucket>();\nconst WINDOW_SIZE_MS = 60_000; \/\/ 1 minute\n\nfunction recordClick(event: any) {\n const campaign = event.data.campaign || 'unknown';\n const key = `${campaign}`;\n\n const bucket = windows.get(key);\n if (bucket && Date.now() - bucket.firstSeen < WINDOW_SIZE_MS) {\n bucket.count++;\n } else {\n \/\/ Emit the completed window\n if (bucket) emitWindow(key, bucket);\n windows.set(key, { campaign, count: 1, firstSeen: Date.now() });\n }\n}\n\nfunction emitWindow(key: string, bucket: WindowBucket) {\n console.log(`Campaign ${bucket.campaign}: ${bucket.count} clicks in window`);\n \/\/ Send to dashboard, check thresholds, etc.\n}\n\n\/\/ Flush windows periodically\nsetInterval(() => {\n const now = Date.now();\n for (const [key, bucket] of windows.entries()) {\n if (now - bucket.firstSeen >= WINDOW_SIZE_MS) {\n emitWindow(key, bucket);\n windows.delete(key);\n }\n }\n}, 10_000);\n<\/code><\/pre>\n\n\n\nExactly-Once Processing<\/h2>\n\n\n\n
const eventHash = crypto\n .createHash('sha256')\n .update(JSON.stringify(event))\n .digest('hex');\n\nawait db.query(\n 'INSERT INTO events (hash, event_type, data) VALUES ($1, $2, $3) ON CONFLICT (hash) DO NOTHING',\n [eventHash, event.event, JSON.stringify(event)]\n);\n<\/code><\/pre>\n\n\n\nasync function processOnce(event: any, handler: () => Promise<void>) {\n const hash = crypto\n .createHash('sha256')\n .update(JSON.stringify(event))\n .digest('hex');\n\n const exists = await db.query(\n 'SELECT 1 FROM processed_events WHERE hash = $1',\n [hash]\n );\n\n if (exists.rows.length > 0) return;\n\n await handler();\n await db.query('INSERT INTO processed_events (hash) VALUES ($1)', [hash]);\n}\n<\/code><\/pre>\n\n\n\nScaling Considerations<\/h2>\n\n\n\n