Edge Computing: Processing Data Close to Users

What is Edge Computing?

Edge computing is a distributed computing paradigm that brings computation and data storage physically closer to the end user — or more precisely, to the "edge" of the network where data is generated and consumed. Instead of routing all requests back to a centralized cloud data center hundreds or thousands of miles away, edge computing processes data on servers located geographically near the user, often within the same metropolitan area or even on the same premises.

The Core Concept

Think of edge computing as placing mini data centers at the "edge" of the internet — in cell towers, retail stores, factory floors, smart vehicles, or regional Points of Presence (PoPs). When a user in Tokyo makes a request, instead of traversing undersea cables to a data center in Virginia, their request is handled by an edge node in Tokyo. The result? Drastically reduced round-trip time and a snappier user experience.

The architecture typically follows this pattern:

Origin Server: The central cloud or on-premises data center that holds the canonical data and runs heavy backend logic
Edge Nodes / PoPs: Geographically distributed servers that cache content, run lightweight compute, and pre-process data
Client Devices: Mobile phones, IoT sensors, browsers, embedded systems that interact primarily with the nearest edge node

Edge vs. Cloud vs. On-Premises

The key distinctions between deployment models are about where computation happens and how far data must travel:

Cloud Computing: Centralized mega data centers (AWS us-east-1, Google us-central1). Massive scale, high latency for distant users, excellent for batch processing and storage
On-Premises: Servers inside the organization's physical location. Full control, zero network latency internally, but limited scalability and high capital expense
Edge Computing: Distributed micro-data centers near users. Low latency, reduced bandwidth costs, works offline, but requires coordination across many nodes

A modern application often uses all three: edge nodes handle real-time requests, cloud handles training ML models, and on-premises systems run legacy ERP workloads — all working in concert.

Why Edge Computing Matters

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

Latency Reduction

The most compelling reason to adopt edge computing is latency. Human perception studies show that a response delay under 100ms feels instantaneous; delays above 300ms are noticeable and irritating. For a user in Mumbai accessing a cloud server in Oregon, network latency alone can exceed 200ms due to the speed-of-light limitations across fiber. An edge node in Mumbai cuts that to under 10ms.

For applications like multiplayer gaming, autonomous vehicle control, or AR/VR experiences, sub-10ms latency isn't just nice — it's a hard requirement. Edge computing makes this achievable.

Bandwidth Optimization

Processing data at the edge means you send only meaningful, aggregated results back to the cloud rather than raw torrents of sensor data. A factory with 10,000 IoT sensors generating 1MB each per second would need a 10GB/s uplink to ship everything to the cloud. An edge gateway that pre-processes and filters that data down to 50KB/s of anomalies and summaries reduces bandwidth costs by 99.5%.

Offline / Intermittent Connectivity

Edge nodes can operate autonomously when the connection to the central cloud is degraded or severed. A point-of-sale system in a rural store, a ship at sea, or a mining operation in a remote desert can continue processing transactions locally and synchronize when connectivity returns. This resilience is impossible with a purely cloud-dependent architecture.

Data Privacy & Sovereignty

Regulations like GDPR, CCPA, and various national data sovereignty laws mandate that certain data must not leave a geographic jurisdiction. Edge computing allows you to process sensitive data (personally identifiable information, health records, financial transactions) within the required legal boundary, sending only anonymized or aggregated results to a global cloud. This keeps you compliant while still benefiting from centralized analytics.

Real-World Use Cases

Content Delivery Networks (CDNs): The original edge computing — cache HTML, images, and video at PoPs worldwide
IoT / Industrial Telemetry: Filter, aggregate, and analyze sensor data on local gateways before cloud ingestion
Real-Time AI Inference: Run lightweight ML models (object detection, speech recognition) on edge servers near cameras or microphones
Gaming & Streaming: Low-latency matchmaking servers and streaming ingest points close to players
Retail & POS Systems: Process transactions, loyalty lookups, and inventory queries locally even during ISP outages
Autonomous Vehicles: Vehicles process LIDAR/camera data onboard (the ultimate edge) while syncing learned patterns to cloud

How to Use Edge Computing — Practical Implementation

Let's walk through concrete code examples that demonstrate edge computing patterns. We'll cover running your own edge server, using serverless edge functions from major providers, handling data at the edge, and processing IoT streams.

Setting Up a Simple Edge Server with Node.js

A basic edge server is essentially a lightweight HTTP server that sits close to users and can handle requests independently. Here's a complete example that demonstrates caching, local processing, and fallback to an origin server:

// edge-server.js — A lightweight edge node implementation
const express = require('express');
const axios = require('axios');
const NodeCache = require('node-cache');
const winston = require('winston');

const app = express();
const PORT = process.env.PORT || 8080;

// In-memory cache with 60-second TTL, check every 120 seconds for stale entries
const cache = new NodeCache({ stdTTL: 60, checkperiod: 120 });

// Logger for edge telemetry
const logger = winston.createLogger({
  level: 'info',
  format: winston.format.json(),
  transports: [
    new winston.transports.File({ filename: '/var/log/edge-node/requests.log' })
  ]
});

// Configuration: where is our origin cloud server?
const ORIGIN_URL = process.env.ORIGIN_URL || 'https://api.mycloudapp.com';
const REGION = process.env.REGION || 'us-east';

// ──────────────────────────────────────────────
// Edge-optimized endpoint: User profile lookup
// ──────────────────────────────────────────────
app.get('/api/users/:userId', async (req, res) => {
  const { userId } = req.params;
  const cacheKey = `user_profile_${userId}`;

  // 1. Check edge cache first — sub-millisecond response if hit
  const cached = cache.get(cacheKey);
  if (cached) {
    logger.info({ event: 'cache_hit', userId, region: REGION });
    res.setHeader('X-Edge-Cache', 'HIT');
    res.setHeader('X-Edge-Region', REGION);
    return res.json(cached);
  }

  // 2. Cache miss — fetch from origin cloud
  try {
    const originResponse = await axios.get(`${ORIGIN_URL}/api/users/${userId}`, {
      timeout: 3000,
      headers: { 'X-Forwarded-Edge': REGION }
    });

    const userData = originResponse.data;

    // 3. Store in edge cache for subsequent requests
    cache.set(cacheKey, userData);
    logger.info({ event: 'origin_fetch', userId, latency: originResponse.headers['x-response-time'] });

    res.setHeader('X-Edge-Cache', 'MISS');
    res.setHeader('X-Edge-Region', REGION);
    return res.json(userData);

  } catch (error) {
    // 4. If origin is unreachable, serve stale data from cache if available
    const staleData = cache.get(cacheKey);
    if (staleData) {
      logger.warn({ event: 'serving_stale', userId, reason: error.message });
      res.setHeader('X-Edge-Stale', 'true');
      res.setHeader('X-Edge-Region', REGION);
      return res.json(staleData);
    }

    // 5. Complete failure — return a graceful degradation response
    logger.error({ event: 'complete_failure', userId, error: error.message });
    res.status(503).json({
      error: 'Service temporarily unavailable',
      retryAfter: 5,
      region: REGION
    });
  }
});

// ──────────────────────────────────────────────
// Health check endpoint for load balancers
// ──────────────────────────────────────────────
app.get('/health', (req, res) => {
  res.json({
    status: 'healthy',
    region: REGION,
    uptime: process.uptime(),
    cacheStats: cache.getStats()
  });
});

app.listen(PORT, () => {
  console.log(`Edge node [${REGION}] listening on port ${PORT}`);
});

This edge server demonstrates several critical patterns: cache-first reads, graceful fallback to stale data when the origin is unreachable, region tagging for observability, and structured logging that can be shipped to a central analytics system asynchronously.

Deploying to Cloudflare Workers (Edge Functions)

Cloudflare Workers run on Cloudflare's global network of over 300 cities. Your code is deployed to every PoP and executes at the point closest to the user. Here's a complete worker that modifies HTML responses on the fly — injecting geo-specific content at the edge:

// workers-site/index.js — Cloudflare Worker for edge personalization
// Deploy with: wrangler publish

addEventListener('fetch', event => {
  event.respondWith(handleRequest(event.request));
});

/**
 * Main edge handler — intercepts requests and personalizes responses
 * based on the visitor's geographic location (derived at the edge).
 */
async function handleRequest(request) {
  // Cloudflare provides geolocation data via request properties
  const cf = request.cf || {};
  const country = cf.country || 'unknown';
  const city = cf.city || 'unknown';
  const timezone = cf.timezone || 'UTC';

  // Fetch from origin server
  const originResponse = await fetch(request);

  // Only transform HTML responses at the edge
  const contentType = originResponse.headers.get('content-type') || '';
  if (!contentType.includes('text/html')) {
    // Pass through non-HTML responses unchanged (images, JS, CSS, etc.)
    return originResponse;
  }

  // Read the HTML body
  let html = await originResponse.text();

  // ──────────────────────────────────────────
  // Edge personalization: inject location-specific content
  // ──────────────────────────────────────────
  const localGreeting = getLocalGreeting(country, timezone);
  const localOffer = getRegionalOffer(country);

  // Replace placeholders in the HTML with edge-computed values
  html = html.replace('{{GREETING}}', localGreeting);
  html = html.replace('{{REGIONAL_OFFER}}', localOffer);
  html = html.replace('{{CITY_NAME}}', city);

  // Insert edge metadata as an invisible comment for debugging
  const edgeMetadata = `\n`;
  html = html.replace('');

  // Return the modified response with appropriate headers
  const modifiedResponse = new Response(html, {
    status: originResponse.status,
    statusText: originResponse.statusText,
    headers: originResponse.headers
  });

  // Add edge-specific headers
  modifiedResponse.headers.set('X-Edge-Country', country);
  modifiedResponse.headers.set('X-Edge-City', city);
  modifiedResponse.headers.set('X-Edge-Processed-By', cf.colo || 'unknown');

  return modifiedResponse;
}

/**
 * Generates a time-of-day and country-appropriate greeting.
 * This logic runs AT THE EDGE, not on the origin server.
 */
function getLocalGreeting(country, timezone) {
  // Calculate local time based on timezone offset
  const now = new Date();
  // Simplified: use UTC hour as approximation
  const hour = now.getUTCHours();

  if (hour >= 5 && hour < 12) return `Good morning from ${country}!`;
  if (hour >= 12 && hour < 17) return `Good afternoon from ${country}!`;
  if (hour >= 17 && hour < 21) return `Good evening from ${country}!`;
  return `Hello from ${country} — it's late, grab some sleep!`;
}

/**
 * Returns a regional promotion based on the visitor's country.
 * In production, this would query an edge-local KV store.
 */
function getRegionalOffer(country) {
  const offers = {
    'US': '🇺🇸 Free shipping on orders over $50!',
    'GB': '🇬🇧 Next-day delivery available in London metro area.',
    'JP': '🇯🇵 TEXTCAMPAIGNTEXT！',
    'DE': '🇩🇪 Kostenloser Versand innerhalb Deutschlands.',
    'BR': '🇧🇷 Frete grátis para as regiões Sul e Sudeste!',
    'AU': '🇦🇺 Express shipping to Sydney and Melbourne available.',
  };
  return offers[country] || '🌍 Worldwide shipping available — order today!';
}

The worker showcases geolocation-based personalization without any client-side JavaScript or origin-server overhead. The country, city, and data center information are provided natively by Cloudflare's runtime — no external API calls needed.

Using AWS Lambda@Edge

Lambda@Edge runs your code at AWS CloudFront edge locations. Unlike Cloudflare Workers which intercept every request, Lambda@Edge hooks into specific CloudFront lifecycle events. Here's a complete example that implements A/B testing at the edge by routing traffic based on viewer cookies:

// lambda-edge-ab-test.js — AWS Lambda@Edge for A/B test routing
// Attach to CloudFront as a Viewer Request trigger
// Runtime: Node.js 18.x

exports.handler = async (event, context) => {
  // CloudFront event structure
  const request = event.Records[0].cf.request;
  const headers = request.headers;

  // ──────────────────────────────────────────
  // Step 1: Check for existing A/B test cookie
  // ──────────────────────────────────────────
  let experimentBucket = null;

  const cookieHeader = headers.cookie
    ? (Array.isArray(headers.cookie)
        ? headers.cookie.map(c => c.value).join('; ')
        : headers.cookie)
    : '';

  if (cookieHeader.includes('ab_experiment=')) {
    // Extract existing bucket assignment from cookie
    const match = cookieHeader.match(/ab_experiment=([^;]+)/);
    if (match) {
      experimentBucket = match[1];
    }
  }

  // ──────────────────────────────────────────
  // Step 2: If no cookie, randomly assign a bucket
  // ──────────────────────────────────────────
  if (!experimentBucket) {
    // Deterministic hashing based on viewer IP for consistent assignment
    const viewerIp = headers['client-ip']
      ? headers['client-ip'][0].value
      : '0.0.0.0';

    // Simple hash to get a number between 0 and 99
    const hashValue = viewerIp.split('.').reduce((acc, octet) => {
      return acc + parseInt(octet, 10);
    }, 0) % 100;

    if (hashValue < 50) {
      experimentBucket = 'variant_a';  // Control group — 50%
    } else if (hashValue < 75) {
      experimentBucket = 'variant_b';  // Treatment 1 — 25%
    } else {
      experimentBucket = 'variant_c';  // Treatment 2 — 25%
    }
  }

  // ──────────────────────────────────────────
  // Step 3: Route request to the appropriate origin path
  // ──────────────────────────────────────────
  // Modify the origin path based on experiment bucket
  const originalUri = request.uri;

  if (experimentBucket === 'variant_b') {
    // Route to a different origin path that serves the B variant
    request.uri = `/experiment-b${originalUri}`;
  } else if (experimentBucket === 'variant_c') {
    request.uri = `/experiment-c${originalUri}`;
  }
  // variant_a uses the default (unchanged) path

  // ──────────────────────────────────────────
  // Step 4: Set/update the cookie via response headers
  // (Note: Lambda@Edge viewer request can only modify the request;
  //  to set cookies, pair this with a viewer response trigger)
  // ──────────────────────────────────────────
  // Store bucket assignment in request headers for the origin to echo back as a cookie
  request.headers['x-ab-experiment'] = [
    {
      key: 'X-AB-Experiment',
      value: experimentBucket
    }
  ];

  // Log experiment assignment (logs appear in CloudWatch in the edge region)
  console.log(JSON.stringify({
    event: 'ab_assignment',
    bucket: experimentBucket,
    uri: request.uri,
    viewerIp: headers['client-ip'] ? headers['client-ip'][0].value : 'unknown',
    timestamp: new Date().toISOString()
  }));

  // Return the modified request to CloudFront for processing
  return request;
};

This Lambda@Edge function demonstrates deterministic user bucketing for A/B tests, ensuring the same user always sees the same variant without needing a central database lookup. The hashing based on client IP provides consistent assignment while operating entirely at the edge.

Vercel Edge Functions

Vercel Edge Functions are deployed to Vercel's global edge network and are ideal for Next.js applications. They support both geolocation and streaming responses. Here's a complete edge function that implements a real-time visitor counter with geo-distributed aggregation:

// app/api/visitor-counter/route.ts — Vercel Edge Function (Next.js App Router)
// File location: app/api/visitor-counter/route.ts

import { NextRequest, NextResponse } from 'next/server';
import { geolocation } from '@vercel/edge';

// Edge config — ensures this runs on Vercel's edge network
export const runtime = 'edge';

/**
 * GET /api/visitor-counter
 * Returns current visitor count for the user's geographic region.
 * Demonstrates edge-native KV storage via Vercel KV (Edge KV).
 */
export async function GET(request: NextRequest) {
  // ──────────────────────────────────────
  // 1. Extract geolocation at the edge
  // ──────────────────────────────────────
  const geo = geolocation(request);
  const country = geo.country || 'unknown';
  const city = geo.city || 'unknown';
  const region = geo.region || 'unknown';

  // ──────────────────────────────────────
  // 2. Get client fingerprint for unique visitor counting
  // ──────────────────────────────────────
  const forwardedFor = request.headers.get('x-forwarded-for') || '0.0.0.0';
  const userAgent = request.headers.get('user-agent') || '';
  const visitorFingerprint = `${forwardedFor}_${userAgent.slice(0, 40)}`;

  // ──────────────────────────────────────
  // 3. Edge KV operations for counting
  // ──────────────────────────────────────
  // In production, use Vercel KV or a similar edge-compatible store
  // This demonstrates the pattern with a simulated edge store
  const kv = getEdgeKVStore();

  // Increment total counter for this geographic region
  const regionKey = `visitors:region:${country}:${region}`;
  const regionCount = await kv.incr(regionKey);

  // Check if this specific visitor was counted in the last hour
  const visitorKey = `visitors:unique:${visitorFingerprint}:${Date.now()}`;
  const isDuplicate = await kv.exists(visitorKey);

  if (!isDuplicate) {
    // Mark this visitor as counted (expire after 1 hour)
    await kv.set(visitorKey, '1', { ttl: 3600 });

    // Increment unique visitor count for the region
    const uniqueKey = `visitors:unique:${country}:${region}:${getCurrentHour()}`;
    await kv.incr(uniqueKey);
  }

  // ──────────────────────────────────────
  // 4. Fetch global aggregated counts
  // ──────────────────────────────────────
  const globalCount = await kv.get('visitors:global:total') || 0;
  const regionUniqueCount = await kv.get(
    `visitors:unique:${country}:${region}:${getCurrentHour()}`
  ) || 0;

  // ──────────────────────────────────────
  // 5. Return response with edge metadata
  // ──────────────────────────────────────
  return NextResponse.json({
    region: {
      country,
      city,
      region,
      total: regionCount,
      uniqueThisHour: regionUniqueCount,
    },
    global: {
      total: globalCount,
    },
    edge: {
      // Which edge data center handled this request
      colo: request.headers.get('x-vercel-ip-country') || 'unknown',
      timestamp: new Date().toISOString(),
    },
  }, {
    headers: {
      'X-Edge-Region': `${country}-${region}`,
      'Cache-Control': 'public, max-age=30, s-maxage=30',
    },
  });
}

// ──────────────────────────────────────────────
// Simulated Edge KV store (replace with Vercel KV in production)
// ──────────────────────────────────────────────
interface EdgeKVStore {
  incr(key: string): Promise;
  get(key: string): Promise;
  set(key: string, value: string, opts?: { ttl: number }): Promise;
  exists(key: string): Promise;
}

function getEdgeKVStore(): EdgeKVStore {
  // In a real Vercel Edge Function, you'd use:
  // import { kv } from '@vercel/kv';
  // For this example, we use an in-memory Map (resets on cold starts)
  const store = new Map();

  return {
    async incr(key: string): Promise {
      const existing = store.get(key);
      const currentValue = existing ? parseInt(existing.value, 10) : 0;
      const newValue = currentValue + 1;
      store.set(key, { value: String(newValue) });
      return newValue;
    },
    async get(key: string): Promise {
      const entry = store.get(key);
      if (!entry) return null;
      if (entry.expiresAt && Date.now() > entry.expiresAt) {
        store.delete(key);
        return null;
      }
      return parseInt(entry.value, 10);
    },
    async set(key: string, value: string, opts?: { ttl: number }): Promise {
      store.set(key, {
        value,
        expiresAt: opts?.ttl ? Date.now() + opts.ttl * 1000 : undefined,
      });
    },
    async exists(key: string): Promise {
      const entry = store.get(key);
      if (!entry) return false;
      if (entry.expiresAt && Date.now() > entry.expiresAt) {
        store.delete(key);
        return false;
      }
      return true;
    },
  };
}

function getCurrentHour(): string {
  return Math.floor(Date.now() / 3600000).toString();
}

This edge function demonstrates geo-aware visitor counting, deduplication logic running entirely at the edge, and edge-native storage patterns. The simulated KV store illustrates the interface you'd use with real edge storage services like Vercel KV, Cloudflare KV, or Redis Enterprise.

Edge Database Strategies — SQLite at the Edge

Running a full database at the edge is now possible with SQLite-based solutions like Turso, Cloudflare D1, or custom SQLite replicas. Here's a complete pattern for synchronizing edge-local SQLite data with a central PostgreSQL database:

// edge-db-sync.js — Bidirectional sync between edge SQLite and cloud PostgreSQL
// This runs on each edge node to maintain local data autonomy

const Database = require('better-sqlite3');
const { Pool } = require('pg');
const cron = require('node-cron');

// ──────────────────────────────────────
// Edge-local SQLite database
// ──────────────────────────────────────
const edgeDB = new Database('/data/edge-local.db');
edgeDB.pragma('journal_mode = WAL');
edgeDB.pragma('synchronous = NORMAL');

// Initialize edge-local schema
edgeDB.exec(`
  CREATE TABLE IF NOT EXISTS transactions (
    id TEXT PRIMARY KEY,
    amount REAL NOT NULL,
    customer_id TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    synced_to_cloud INTEGER NOT NULL DEFAULT 0
  );

  CREATE TABLE IF NOT EXISTS product_inventory (
    sku TEXT PRIMARY KEY,
    quantity INTEGER NOT NULL,
    price REAL NOT NULL,
    last_synced TEXT
  );

  CREATE INDEX IF NOT EXISTS idx_transactions_synced
    ON transactions(synced_to_cloud);

  CREATE INDEX IF NOT EXISTS idx_transactions_created
    ON transactions(created_at);
`);

// ──────────────────────────────────────
// Cloud PostgreSQL connection pool
// ──────────────────────────────────────
const cloudPool = new Pool({
  host: process.env.CLOUD_DB_HOST || 'pg-primary.mycloud.com',
  port: parseInt(process.env.CLOUD_DB_PORT || '5432'),
  database: process.env.CLOUD_DB_NAME || 'main',
  user: process.env.CLOUD_DB_USER || 'edge_sync_user',
  password: process.env.CLOUD_DB_PASSWORD,
  max: 5,
  connectionTimeoutMillis: 10000,
});

// ──────────────────────────────────────
// UPLOAD: Push local transactions to cloud
// ──────────────────────────────────────
async function uploadLocalTransactions() {
  const unsyncedTransactions = edgeDB.prepare(`
    SELECT id, amount, customer_id, status, created_at
    FROM transactions
    WHERE synced_to_cloud = 0
    ORDER BY created_at ASC
    LIMIT 100
  `).all();

  if (unsyncedTransactions.length === 0) {
    console.log('[sync] No pending transactions to upload');
    return;
  }

  console.log(`[sync] Uploading ${unsyncedTransactions.length} transactions to cloud`);

  try {
    const client = await cloudPool.connect();

    try {
      await client.query('BEGIN');

      for (const tx of unsyncedTransactions) {
        await client.query(
          `INSERT INTO transactions (id, amount, customer_id, status, created_at, edge_node_id)
           VALUES ($1, $2, $3, $4, $5, $6)
           ON CONFLICT (id) DO UPDATE SET
             amount = EXCLUDED.amount,
             status = EXCLUDED.status`,
          [tx.id, tx.amount, tx.customer_id, tx.status, tx.created_at, process.env.EDGE_NODE_ID]
        );
      }

      await client.query('COMMIT');

      // Mark as synced in local SQLite
      const markSynced = edgeDB.prepare(`
        UPDATE transactions SET synced_to_cloud = 1
        WHERE id = ?
      `);

      const markMany = edgeDB.transaction((ids) => {
        for (const id of ids) {
          markSynced.run(id);
        }
      });

      markMany(unsyncedTransactions.map(tx => tx.id));
      console.log(`[sync] Successfully uploaded and marked ${unsyncedTransactions.length} transactions`);

    } finally {
      client.release();
    }
  } catch (error) {
    console.error('[sync] Upload failed:', error.message);
    // Transactions remain marked synced_to_cloud=0 for retry on next cycle
  }
}

// ──────────────────────────────────────
// DOWNLOAD: Pull inventory updates from cloud
// ──────────────────────────────────────
async function downloadInventoryUpdates() {
  try {
    const client = await cloudPool.connect();

    try {
      // Fetch inventory changes since last sync
      const lastSyncTime = edgeDB.prepare(`
        SELECT MAX(last_synced) as max_sync FROM product_inventory
      `).get()?.max_sync || '1970-01-01T00:00:00Z';

      const result = await client.query(
        `SELECT sku, quantity, price, updated_at
         FROM product_inventory
         WHERE updated_at > $1
         ORDER BY updated_at ASC`,
        [lastSyncTime]
      );

      if (result.rows.length === 0) {
        return;
      }

      console.log(`[sync] Downloading ${result.rows.length} inventory updates`);

      const upsert = edgeDB.prepare(`
        INSERT INTO product_inventory (sku, quantity, price, last_synced)
        VALUES (?, ?, ?, ?)
        ON CONFLICT(sku) DO UPDATE SET
          quantity = excluded.quantity,
          price = excluded.price,
          last_synced = excluded.last_synced
      `);

      const upsertAll = edgeDB.transaction((rows) => {
        for (const row of rows) {
          upsert.run(row.sku, row.quantity, row.price, row.updated_at);
        }
      });

      upsertAll(result.rows);
      console.log(`[sync] Updated ${result.rows.length} inventory records locally`);

    } finally {
      client.release();
    }
  } catch (error) {
    console.error('[sync] Download failed:', error.message);
  }
}

// ──────────────────────────────────────
// Full sync cycle
// ──────────────────────────────────────
async function fullSyncCycle() {
  console.log('[sync] Starting sync cycle...');
  await uploadLocalTransactions();
  await downloadInventoryUpdates();
  console.log('[sync] Sync cycle complete');
}

// Run sync every 30 seconds
cron.schedule('*/30 * * * * *', () => {
  fullSyncCycle().catch(err => console.error('[sync] Cycle error:', err));
});

// Also run on startup
fullSyncCycle();

// Export for use by the edge API server
module.exports = { edgeDB, cloudPool, fullSyncCycle };

This sync module demonstrates bidirectional edge-cloud synchronization with idempotent upserts, batch processing, and transactional marking. The edge node can operate entirely from its local SQLite database even when the cloud connection fails, and automatically catches up when connectivity is restored.

Processing IoT Data at the Edge (Python)

For industrial IoT scenarios, edge processing often runs on gateways using Python. Here's a complete edge processor that filters, aggregates, and forwards sensor data:

# iot_edge_processor.py — Industrial IoT edge gateway processor
# Runs on Raspberry Pi or similar edge hardware at the factory floor
# Requires: pip install paho-mqtt numpy requests

import json
import time
import threading
import queue
from collections import deque
from datetime import datetime, timedelta
import numpy as np
import paho.mqtt.client as mqtt
import requests

# ───────────────────────────────────────
# Configuration
# ───────────────────────────────────────
EDGE_GATEWAY_ID = "gw-floor3-east-01"
MQTT_BROKER = "localhost"  # Local MQTT broker on the gateway
CLOUD_INGEST_URL = "https://api.cloud-iot.mycompany.com/ingest"
CLOUD_BATCH_INTERVAL = 30  # seconds between cloud uploads
ANOMALY_THRESHOLD_STDDEV = 3.0  # Flag readings beyond 3 standard deviations
LOCAL_BUFFER_SIZE = 1000  # Keep last 1000 readings per sensor in memory

# ───────────────────────────────────────
# In-memory sliding window for each sensor
# ───────────────────────────────────────
sensor_buffers = {}  # sensor_id -> deque of (timestamp, value)
sensor_stats = {}    # sensor_id -> {"mean": float, "std": float, "n": int}
anomaly_queue = []   # List of detected anomalies to forward immediately
batch_queue = []     # Aggregated readings for periodic cloud upload

# Thread-safe queue for processing incoming sensor readings
reading_queue = queue.Queue(maxsize=10000)

# Shutdown flag for graceful termination
running = True

# ───────────────────────────────────────
# MQTT Callbacks — ingest raw sensor data
# ───────────────────────────────────────
def on_connect(client, userdata, flags, rc):
    print(f"[MQTT] Connected to local broker with result code {rc}")
    # Subscribe to all sensor topics on the factory floor
    client.subscribe("factory/+/sensors/#")

def on_message(client, userdata, msg):
    """Called for every incoming MQTT message from sensors"""
    try:
        payload = json.loads(msg.payload.decode('utf-8'))
        sensor_id = msg.topic.split('/')[-1]  # Extract sensor ID from topic

        reading = {
            'sensor_id': sensor_id,
            'value': float(payload.get('value', 0)),
            'unit': payload.get('