import sqlite3              # Lightweight embedded database that persists across power loss
import json                 # Serialize telemetry dicts to strings for SQLite storage
import time                 # Unix timestamp for ordering records chronologically
from typing import Dict, List  # Type hints for cleaner, self-documenting code
import requests             # HTTP client to POST batches to the regional gateway

class TelemetryBuffer:
    def __init__(self, db_path: str, max_retries: int = 5):
        self.conn = sqlite3.connect(db_path)    # Open (or create) a persistent local SQLite database
        self.max_retries = max_retries          # Upper limit on retry attempts before a record is discarded
        self._init_db()                         # Ensure the telemetry table exists on construction

    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS telemetry (
                id INTEGER PRIMARY KEY AUTOINCREMENT,   # Monotonically increasing row ID for ordering and deletion
                payload TEXT NOT NULL,                   # JSON-serialized telemetry data stored as text
                created_at REAL NOT NULL,                # Unix timestamp for ordering and TTL-based expiry
                retries INTEGER DEFAULT 0               # Tracks how many times this row has failed to send
            )
        """)
        self.conn.commit()                      # Persist the schema change immediately

    def record(self, data: Dict):
        self.conn.execute(
            "INSERT INTO telemetry (payload, created_at) VALUES (?, ?)",  # Parameterized query prevents SQL injection
            (json.dumps(data), time.time())     # Convert dict to JSON string and stamp with current time
        )
        self.conn.commit()                      # Flush the write to disk so it survives a crash

    def flush(self, endpoint: str, batch_size: int = 100):
        cursor = self.conn.execute(
            "SELECT id, payload FROM telemetry WHERE retries < ? LIMIT ?",  # Only pick rows that haven't exhausted retries
            (self.max_retries, batch_size)      # Cap the batch size to bound memory and HTTP payload
        )
        rows = cursor.fetchall()                # Materialize the selected rows as a list of (id, payload) tuples
        if not rows:                            # Nothing to send; exit early to avoid an unnecessary HTTP call
            return

        batch = [json.loads(r[1]) for r in rows]  # Deserialize each stored JSON string back into a dict
        try:
            resp = requests.post(
                endpoint,                       # Typically the regional gateway URL assigned during device registration
                json={"telemetry": batch},      # Wrap the batch in the expected envelope format
                timeout=10                      # Abort if the gateway doesn't respond within 10 seconds
            )
            if resp.status_code == 200:         # Gateway accepted and published the batch successfully
                ids = tuple(r[0] for r in rows) # Collect the primary keys of successfully sent rows
                self.conn.execute(
                    f"DELETE FROM telemetry WHERE id IN ({','.join('?' * len(ids))})",  # Dynamic parameterized IN clause
                    ids                         # Bind all IDs safely — never interpolate user data into SQL
                )
                self.conn.commit()              # Remove the sent records from local storage
        except requests.RequestException:       # Catch network timeouts, DNS failures, connection resets, etc.
            cursor = self.conn.execute(
                "UPDATE telemetry SET retries = retries + 1 WHERE id IN "
                f"({','.join('?' * len(rows))})",  # Increment retry counter for every row in the failed batch
                tuple(r[0] for r in rows)       # Bind the row IDs to the parameterized query
            )
            self.conn.commit()                  # Persist the updated retry counts so the device doesn't retry infinitely

from fastapi import FastAPI, HTTPException   # FastAPI provides async HTTP with automatic OpenAPI docs
from kafka import KafkaProducer               # Kafka producer client for publishing messages to a topic
import json                                   # Serialize telemetry records to bytes for Kafka

app = FastAPI()                               # Create the FastAPI application instance
producer = KafkaProducer(
    bootstrap_servers=["kafka-cluster:9092"], # Points to the Kafka broker cluster endpoint
    compression_type="gzip",                  # Compress messages on the wire to reduce bandwidth and storage
    linger_ms=10,                             # Wait up to 10ms before sending to batch small messages into larger requests
)

@app.post("/telemetry")
async def ingest(batch: dict):
    if "telemetry" not in batch or not isinstance(batch["telemetry"], list):  # Validate the expected envelope structure
        raise HTTPException(400, "Invalid payload")   # Reject malformed batches early before they reach Kafka

    for record in batch["telemetry"]:                 # Iterate over every telemetry point in the batch
        producer.send("telemetry-raw", json.dumps(record).encode())  # Publish each record individually to the raw topic

    return {"status": "accepted", "count": len(batch["telemetry"])}  # Acknowledge receipt so the device can delete its buffer

@app.get("/health")
async def health():
    return {"status": "ok"}                           # Simple liveness probe for the load balancer

from pyflink.datastream import StreamExecutionEnvironment  # Core Flink streaming context
from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkKafkaProducer  # Kafka source and sink connectors
import json                                                # Parse JSON bytes from Kafka into Python dicts

env = StreamExecutionEnvironment.get_execution_environment()  # Initialize the Flink runtime environment

kafka_consumer = FlinkKafkaConsumer(
    topics="telemetry-raw",                               # Subscribe to the ingestion topic where gateways publish
    properties={"bootstrap.servers": "kafka-cluster:9092", "group.id": "telemetry-processor"},  # Broker address and consumer group for offset tracking
    deserialization_schema=lambda x: json.loads(x.decode())  # Deserialize raw Kafka message bytes into Python dicts
)

stream = env.add_source(kafka_consumer)                   # Wire the Kafka consumer as the stream's data source

deduped = stream.key_by(lambda r: (r["device_id"], r["seq"])) \  # Partition by device+sequence to route duplicates to the same worker
                .process(DeduplicateFunction(ttl_seconds=3600))  # Drop records whose device+seq pair was seen within the last hour

metrics = deduped.filter(lambda r: r["type"] == "metric")  # Route metric records to the time-series pipeline
logs = deduped.filter(lambda r: r["type"] == "log")        # Route log records to the object-storage pipeline

metrics.sink_to(TSDB_sink)                                 # Write metrics to TimescaleDB / VictoriaMetrics for dashboard queries
logs.sink_to(S3_sink)                                      # Write logs to S3 for long-term archival and ML training

env.execute("telemetry-processor")                         # Deploy the pipeline to the Flink cluster

import aioredis                               # Async Redis client compatible with FastAPI's event loop
from fastapi import FastAPI, HTTPException    # HTTP framework and error response utilities

redis = aioredis.from_url("redis://redis-cluster:6379")  # Connect to the Redis cluster for distributed rate-limiting state

@app.post("/telemetry")
async def ingest(batch: dict, device_id: str = Header(...)):  # Extract device_id from the mandatory HTTP header
    key = f"ratelimit:{device_id}"              # Namespaced Redis key scoped to a single device
    current = await redis.incr(key)             # Atomically increment the request counter for this device
    if current == 1:                            # First request in this window — set the TTL to start the countdown
        await redis.expire(key, 60)             # The counter resets after 60 seconds, giving the device a fresh allowance
    if current > 10:                            # Device has exceeded 10 requests within the 60-second window
        raise HTTPException(429, "Too many requests. Slow down.")  # HTTP 429 tells the device to back off

class CircuitBreaker:
    def __init__(self, tsdb_client, s3_client):
        self.tsdb = tsdb_client                # Client for the primary time-series database
        self.s3 = s3_client                    # Client for the S3-compatible object store fallback
        self.failures = 0                      # Running count of consecutive write failures to the TSDB
        self.threshold = 5                     # Number of consecutive failures before the circuit opens (trips)
        self.open = False                      # Circuit state: False = closed (normal operation), True = open (bypass TSDB)

    def write(self, data: dict):
        if self.open:                          # Circuit is tripped — skip the TSDB entirely
            self._dump_to_s3(data)             # Write straight to S3 to avoid hammering an unhealthy database
            return

        try:
            self.tsdb.write(data)              # Attempt the normal write path to the time-series database
            self.failures = 0                  # Success resets the failure counter (circuit stays closed)
        except Exception:                      # Any exception from the TSDB write is treated as a failure
            self.failures += 1                 # Increment the consecutive failure counter
            if self.failures >= self.threshold:  # Threshold breached — trip the circuit
                self.open = True               # Subsequent writes will bypass the TSDB until the circuit is manually reset
            self._dump_to_s3(data)             # Fallback: persist the data to S3 so nothing is lost

    def _dump_to_s3(self, data: dict):
        self.s3.put_object(
            Bucket="telemetry-backpressure",   # Dedicated bucket for circuit-breaker overflow data
            Key=f"replay/{data['device_id']}/{uuid.uuid4()}.json",  # Unique object key with device path and random UUID to avoid collisions
            Body=json.dumps(data)              # Serialize the data dict to a JSON string for storage
        )

from fastapi import FastAPI, HTTPException   # HTTP framework and exception handling
import jwt                                   # JSON Web Token library for stateless device authentication

app = FastAPI()                              # Create the FastAPI application instance for the registration service

@app.post("/register")
async def register(device_id: str, public_key: str):  # Accept the device's identity and its public key for later signature verification
    token = jwt.encode(
        {"device_id": device_id, "iat": time.time()},  # Embed the device identity and token issuance timestamp
        secret, algorithm="RS256"            # Sign with an asymmetric RS256 key; the device uses the public half for verification
    )
    return {
        "token": token,                      # The device presents this bearer token on every subsequent telemetry POST
        "gateway_url": assign_gateway(device_id),  # DNS-optimized regional gateway for this device's location
        "config": {
            "interval_s": 300,               # How often (in seconds) the device should attempt a flush
            "batch_size": 100,               # How many records to include in each HTTP POST
            "retry_max": 10,                 # Maximum retry attempts before a record is considered dead
            "backoff_base_ms": 200           # Base delay in milliseconds for exponential back-off between retries
        }
    }