/**
 *  Home Connect Stream Driver v3
 *
 *  Copyright 2026 Craig Dewar
 *
 *  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
 *  in compliance with the License. You may obtain a copy of the License at:
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
 *  on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License
 *  for the specific language governing permissions and limitations under the License.
 *
 *  ===========================================================================================================
 *  ARCHITECTURE OVERVIEW
 *  ===========================================================================================================
 *  
 *  This driver serves as the central hub for Home Connect communication:
 *  
 *  1. SSE CONNECTION: Maintains a single Server-Sent Events connection to Home Connect for all appliances.
 *     The connection receives real-time updates (status changes, program progress, events) and routes
 *     them to the appropriate child device drivers via the parent app.
 *  
 *  2. API LIBRARY: Provides HTTP methods (GET/PUT/DELETE) for all Home Connect API calls.
 *     Child drivers don't make API calls directly - they go through the parent app, which
 *     delegates to this driver.
 *  
 *  3. RATE LIMIT MANAGEMENT: Home Connect enforces strict rate limits (1000 calls/day).
 *     This driver tracks rate limiting, implements conservative reconnect timing, and
 *     automatically schedules reconnection when limits expire.
 *  
 *  Event Flow:
 *  -----------
 *  Home Connect SSE → parse() → processEventPayload() → parent.handleApplianceEvent() → child.parseEvent()
 *  
 *  API Call Flow:
 *  --------------
 *  Child Driver → parent.startProgram() → streamDriver.setActiveProgram() → Home Connect API
 *  
 *  ===========================================================================================================
 *
 *  Version History:
 *  ----------------
 *  3.0.0  2026-01-07  Initial v3 architecture with Stream Driver pattern
 *  3.0.1  2026-01-08  Added conservative reconnect logic (5 min delay for normal disconnects)
 *                     Added rate limit detection and auto-recovery scheduling
 *                     Added exponential backoff for failed connections
 *  3.0.2  2026-01-08  Changed to z_setApiUrl for flexible API URL configuration
 *                     Supports both production and simulator APIs via parent app
 *  3.0.3  2026-01-08  Added lastEventReceived timestamp for stream health monitoring
 *                     Added rateLimitRemaining/rateLimitLimit attributes from API headers
 *  3.1.0  2026-01-21  Hardened code with comprehensive error handling and input validation
 *                     Added safe JSON parsing with error handling (prevents crashes on malformed data)
 *                     Added safe type casting with validation (prevents ClassCastException)
 *                     Added bounds checking for all substring operations
 *                     Added null validation in API callbacks and list iterations
 *                     Significantly improved stability and fault tolerance
 *  3.2.0  2026-01-21  Fixed "Too many follow-up requests" ProtocolException error
 *                     Added duplicate eventStreamStatus() call detection (prevents double-scheduling)
 *                     Added unschedule() calls to prevent overlapping reconnect timers
 *                     Added connection state guards to prevent concurrent connection attempts
 *                     Added specific handler for ProtocolException with 60s backoff and token refresh
 *                     Fixed bug where ERROR+STOP events caused 60s reconnect loop
 *  3.2.1  2026-01-21  Added troubleshooting instrumentation for ProtocolException
 *                     Logs OAuth token age and validity at connection time
 *                     Logs connection duration when ProtocolException occurs
 *                     Logs API endpoint URL being used
 *                     Helps identify root cause of redirect loops
 *  3.2.2  2026-01-21  Fixed "No active program" error appearing during app updates
 *                     Changed 404 handling for programs/active to DEBUG instead of throwing exception
 *                     Prevents alarming ERROR logs when dishwasher is idle during initialization
 *  3.2.3  2026-01-21  Fixed driver version not updating after code changes
 *                     Added version update to initialize() and refresh() methods
 *                     Users can now click Initialize or Refresh to update displayed version
 *  3.2.4  2026-01-21  Changed HTTP 409 (Conflict) log level from WARN to DEBUG
 *                     409 responses are expected when appliance is not ready (door open, transitioning)
 *                     Reduces log noise for normal operating conditions
 *  3.2.5  2026-01-21  Added defensive try/catch for getTokenExpiryTime() calls
 *                     Prevents error if parent app hasn't been updated to v3.1.1 yet
 *                     Stream Driver now gracefully handles missing parent app method
 *  3.2.6  2026-01-23  Improved SSE event diagnostics and stream management
 *                     Added event statistics tracking to diagnose stream issues
 *                     Added reconnect command to force stream refresh when events stop
 *                     Added showEventStats/clearEventStats commands for troubleshooting
 *                     Changed event routing logs from DEBUG to INFO for better visibility
 *                     Added logging when stream has no items in event payload
 *  3.2.7  2026-01-23  Improved error handling for 409 Conflict and 503 Service Unavailable
 *                     Added user-friendly error messages when commands fail
 *                     Automatically updates device lastCommandStatus with error details
 *                     Extracts specific reason from error responses (door open, remote control, etc.)
 *                     Changed 409 error log level from DEBUG to INFO for visibility
 *  3.2.8  2026-02-03  Improved 409 error messages with user-friendly translations
 *                     Translates API error keys (UnsupportedSetting, UnsupportedOption, etc.)
 *                     Extracts setting/option name from API path for context in messages
 *                     Changed 409 command rejection log level to WARN for visibility
 *                     Messages now clearly indicate what is not supported by the appliance
 *  3.2.9  2026-02-20  Fixed double-processing of SSE events in parse() method
 *                     Removed redundant single-line data: handler that could cause
 *                     duplicate event delivery alongside the message buffer processing
 *  3.3.0  2026-02-20  Added stream health watchdog to detect silently dead connections
 *                     Home Connect sends KEEP-ALIVE every ~55s; if no data arrives for
 *                     3 minutes, the watchdog forces a reconnect automatically
 *                     Fixes issue where SSE connection drops silently and events stop
 *                     Added message buffer overflow protection (10KB limit)
 *  3.3.1  2026-02-20  Auto-initialize on driver code update
 *                     updated() now detects version change and calls initialize()
 *                     Eliminates need to manually click Initialize after HPM update
 *                     Always refresh device status on reconnect (removed 5-min threshold)
 *                     Fetch active program on reconnect for complete state recovery
 *  3.3.2  2026-02-20  Added verbose stream diagnostics toggle for debugging stale events
 *                     Tracks per-appliance last event times to identify which appliance stops
 *                     Watchdog now logs detailed stream health: KEEP-ALIVE vs real event gaps
 *                     Added showStreamHealth command for on-demand diagnostics
 *  3.3.15 2026-02-20  Fix watchdog failing to detect dead stream when state.lastParseTime
 *                     is null (Hubitat state loss). Falls back to lastRealEventTime or
 *                     connectionStartTime so the timeout check always works.
 *                     Fix SSE buffer not processing messages: Hubitat strips newlines from
 *                     parse() chunks and empty calls represent blank-line delimiters.
 *                     Re-add line terminators so \n\n boundaries form and events reach
 *                     child devices.
 *  3.3.16 2026-02-20  Handle bare JSON lines in SSE messages (Hubitat may deliver data
 *                     without "data:" prefix). Fallback parsing ensures events are not
 *                     silently dropped when the standard "data:" field is missing.
 *  3.3.18 2026-02-21  Fix keep-alive detection: Home Connect sends keep-alives as bare
 *                     "data:" (no event: field, no JSON payload). Previous code required
 *                     line.length() > 5 which silently dropped them. Now detected via
 *                     hasDataField with no payload. KEEP-ALIVE counter works correctly.
 *  3.3.21 2026-03-12  Fixed connectionStatus showing "disconnected" while events still flowing.
 *                     Hubitat can fire eventStreamStatus(STOP) while the SSE connection is still
 *                     alive. parse() now corrects stale connectionStatus back to "connected" when
 *                     data arrives, preventing false watchdog reconnects and status confusion.
 *  3.3.20 2026-03-12  Fixed disconnect() killing the persistent watchdog cron schedule.
 *                     Fixed refresh()/reconnect() race condition: two near-simultaneous calls
 *                     could create duplicate SSE connections via pauseExecution(). Now uses
 *                     runIn() which deduplicates automatically.
 *  3.3.19 2026-03-12  Fix SSE stream not auto-recovering after connection loss.
 *                     Watchdog now uses persistent cron schedule instead of fragile runIn() chain.
 *                     Watchdog now auto-recovers from ALL non-connected states (was only "connected").
 *                     Fixed connect() being blocked by stale "connecting" status (60s timeout added).
 *                     Fixed processingDisconnect flag getting stuck (cleared on every connect attempt).
 *                     Child device "Initialize and Configure" now also reconnects broken SSE stream.
 *                     Resolves issue where all appliance status updates stop after stream disconnect.
 *  3.3.17 2026-02-21  Add self-tracked API call counter (apiCallsToday attribute) with
 *                     per-method/category breakdown.  Counter resets are synced to Home
 *                     Connect's own window via X-RateLimit-Remaining header (when it jumps
 *                     up, their window reset, so we reset ours).  Warnings driven by the
 *                     API-reported remaining count.  showApiUsage/resetApiCounter commands.
 *                     Auto-detect driver code updates (HPM) via checkDriverVersion() in
 *                     parse() — triggers re-initialize on first event after an update.
 */

import groovy.json.JsonSlurper
import groovy.json.JsonOutput
import groovy.transform.Field

// Buffer for accumulating SSE message fragments across multiple parse() calls
@Field static String messageBuffer = ""

metadata {
    definition(name: "Home Connect Stream Driver v3", namespace: "craigde", author: "Craig Dewar") {
        capability "Initialize"
        capability "Refresh"

        // User-facing commands
        command "connect"
        command "disconnect"
        command "clearRateLimit"
        command "reconnect", [[name: "Reconnect", type: "STRING", description: "Disconnect and reconnect to refresh stream"]]
        command "showEventStats", [[name: "Show Event Statistics", type: "STRING", description: "Display event statistics for diagnostics"]]
        command "clearEventStats", [[name: "Clear Event Statistics", type: "STRING", description: "Clear event statistics"]]
        command "showStreamHealth", [[name: "Show Stream Health", type: "STRING", description: "Show detailed stream health diagnostics"]]
        command "showApiUsage", [[name: "Show API Usage", type: "STRING", description: "Show API call counts for the current 24-hour window"]]
        command "resetApiCounter", [[name: "Reset API Counter", type: "STRING", description: "Reset the API call counter"]]

        // Internal command (z_ prefix convention)
        command "z_deviceLog", [[name: "level", type: "STRING"], [name: "msg", type: "STRING"]]
        command "z_setApiUrl", [[name: "url", type: "STRING"]]

        // Attributes
        attribute "connectionStatus", "string"   // connected, disconnected, connecting, rate limited, error
        attribute "lastEventTime", "string"      // Timestamp of last received event (legacy - formatted)
        attribute "lastEventReceived", "string"  // ISO timestamp of last SSE event for health monitoring
        attribute "rateLimitRemaining", "number" // Remaining API calls (from response headers)
        attribute "rateLimitLimit", "number"     // Total API call limit (from response headers)
        attribute "apiCallsToday", "number"     // Self-tracked API call count for current 24-hour window
        attribute "apiUrl", "string"             // Current API URL being used
        attribute "driverVersion", "string"
    }

    preferences {
        input name: "debugLogging", type: "bool", title: "Enable debug logging", defaultValue: false,
              description: "Enable detailed logging for troubleshooting. Disable for normal operation."
        input name: "verboseStreamDiag", type: "bool", title: "Enable verbose stream diagnostics", defaultValue: false,
              description: "Logs detailed stream health: per-appliance event times, KEEP-ALIVE vs real event gaps. Toggle on to diagnose stale event issues."
    }
}

/* ===========================================================================================================
   CONSTANTS
   =========================================================================================================== */

@Field static final String DEFAULT_API_URL = "https://api.home-connect.com"
@Field static final String ENDPOINT_APPLIANCES = "/api/homeappliances"
@Field static final String DRIVER_VERSION = "3.3.21"

// Reconnect timing constants
@Field static final Integer NORMAL_RECONNECT_DELAY = 300      // 5 minutes after normal disconnect
@Field static final Integer MAX_RECONNECT_ATTEMPTS = 10       // Give up after this many failed attempts
@Field static final Integer RATE_LIMIT_BUFFER = 300           // 5 minute buffer after rate limit expires

// Stream health watchdog constants
// Home Connect sends KEEP-ALIVE every ~55 seconds. If no data arrives for STREAM_TIMEOUT,
// the connection is considered dead and will be automatically reconnected.
@Field static final Integer STREAM_WATCHDOG_INTERVAL = 300    // Check stream health every 5 minutes
@Field static final Integer STREAM_TIMEOUT = 180000           // 3 minutes without data = dead stream

/**
 * Gets the API URL - can be overridden by parent app via z_setApiUrl
 */
private String getApiUrl() {
    return state.apiUrl ?: DEFAULT_API_URL
}

/* ===========================================================================================================
   LIFECYCLE METHODS
   =========================================================================================================== */

def installed() {
    log.info "Home Connect Stream Driver v3 installed"
    sendEvent(name: "driverVersion", value: DRIVER_VERSION)
    sendEvent(name: "connectionStatus", value: "disconnected")
}

def updated() {
    log.info "Home Connect Stream Driver v3 updated"
    def previousVersion = device.currentValue("driverVersion")
    sendEvent(name: "driverVersion", value: DRIVER_VERSION)
    if (previousVersion != DRIVER_VERSION) {
        logInfo("Driver updated from ${previousVersion} to ${DRIVER_VERSION}, re-initializing")
        runIn(1, "initialize")
    }
}

/**
 * Called when device is initialized or hub restarts
 * Automatically attempts to connect to Home Connect
 * Also updates the driver version attribute
 */
def initialize() {
    logInfo("Initializing Home Connect Stream Driver v${DRIVER_VERSION}")
    sendEvent(name: "driverVersion", value: DRIVER_VERSION)

    // Use a persistent cron schedule for the watchdog so it always runs,
    // even if a one-shot runIn() was lost due to hub restart or state corruption.
    // Runs every 5 minutes. This replaces the fragile one-shot runIn() chain.
    unschedule("streamWatchdog")
    schedule("0 */5 * * * ?", "streamWatchdog")

    // Clear any stuck state that might prevent reconnection
    state.processingDisconnect = false

    connect()
}

/**
 * Detects driver code updates (e.g. via HPM) that don't trigger updated().
 * Called from parse() so the first SSE event after an update triggers re-initialization.
 */
private void checkDriverVersion() {
    if (device.currentValue("driverVersion") != DRIVER_VERSION) {
        sendEvent(name: "driverVersion", value: DRIVER_VERSION)
        logInfo("Driver code updated to v${DRIVER_VERSION}, scheduling re-initialization")
        runIn(2, "initialize")
    }
}

/**
 * Refreshes the connection by disconnecting and reconnecting
 * Also updates the driver version attribute
 * Uses runIn() instead of pauseExecution() to prevent double-connect race conditions
 */
def refresh() {
    logInfo("Refreshing connection")
    sendEvent(name: "driverVersion", value: DRIVER_VERSION)
    disconnect()
    runIn(2, "connect")
}

/* ===========================================================================================================
   SSE CONNECTION MANAGEMENT
   =========================================================================================================== */

/**
 * Establishes SSE connection to Home Connect event stream
 * Handles rate limiting, token validation, and connection setup
 *
 * This method cancels any pending reconnection attempts before connecting
 * to prevent overlapping connection attempts from stacking up.
 */
def connect() {
    // Cancel any previously scheduled connect() calls to prevent overlap
    unschedule("connect")

    // Clear stuck processingDisconnect flag (prevents silent ignoring of disconnect events)
    state.processingDisconnect = false

    // Check if we're already connecting or connected
    def currentStatus = device.currentValue("connectionStatus")
    if (currentStatus == "connected") {
        logDebug("Already connected - ignoring duplicate connect() call")
        return
    }
    if (currentStatus == "connecting") {
        // Check if we've been stuck in "connecting" state for too long (> 60 seconds)
        def connectStart = state.lastConnectAttemptTime ?: 0
        if (connectStart > 0 && (now() - connectStart) < 60000) {
            logDebug("Already connecting (${((now() - connectStart) / 1000).toInteger()}s ago) - waiting")
            return
        }
        logWarn("Stuck in 'connecting' state for over 60s - forcing reconnect")
        sendEvent(name: "connectionStatus", value: "disconnected")
    }

    // Check if we're rate limited
    if (state.rateLimitedUntil && now() < state.rateLimitedUntil) {
        def rateLimitTime = state.rateLimitedUntilFormatted ?: formatDateTime(state.rateLimitedUntil)
        logWarn("Cannot connect - rate limited until ${rateLimitTime}")
        sendEvent(name: "connectionStatus", value: "rate limited until ${rateLimitTime}")
        return
    }

    // Clear expired rate limit state
    if (state.rateLimitedUntil && now() >= state.rateLimitedUntil) {
        state.rateLimitedUntil = null
        state.rateLimitedUntilFormatted = null
        state.reconnectAttempts = 0
        logInfo("Rate limit expired - cleared")
    }

    logDebug("Connecting to Home Connect event stream")

    def token = parent?.getOAuthToken()
    if (!token) {
        logError("No OAuth token available - cannot connect")
        sendEvent(name: "connectionStatus", value: "error - no token")
        return
    }

    // Log token metadata for troubleshooting ProtocolException
    try {
        def tokenExpires = parent?.getTokenExpiryTime() ?: 0
        def tokenAge = tokenExpires > 0 ? (tokenExpires - now()) / 1000 : -1
        logDebug("OAuth token valid for ${tokenAge}s more")
    } catch (Exception e) {
        // Parent app may not have getTokenExpiryTime() yet - ignore
    }

    def language = parent?.getLanguage() ?: "en-US"
    def apiUrl = getApiUrl()
    logDebug("Connecting to: ${apiUrl}${ENDPOINT_APPLIANCES}/events")

    // Track connect attempt time for "connecting" state timeout detection
    state.lastConnectAttemptTime = now()

    try {
        interfaces.eventStream.connect(
            "${apiUrl}${ENDPOINT_APPLIANCES}/events",
            [
                rawData: true,
                ignoreSSLIssues: true,
                headers: [
                    'Authorization': "Bearer ${token}",
                    'Accept': 'text/event-stream',
                    'Accept-Language': language
                ]
            ]
        )
        sendEvent(name: "connectionStatus", value: "connecting")

    } catch (Exception e) {
        logError("Failed to connect: ${e.message}")
        sendEvent(name: "connectionStatus", value: "error")
    }
}

/**
 * Closes the SSE connection
 * Cancels any pending reconnection attempts
 */
def disconnect() {
    logInfo("Disconnecting from Home Connect event stream")

    // Cancel any scheduled reconnection attempts
    // NOTE: Do NOT unschedule streamWatchdog here - the persistent cron watchdog
    // must keep running to auto-recover from disconnected states
    unschedule("connect")

    try {
        interfaces.eventStream.close()
    } catch (Exception e) {
        logWarn("Error during disconnect: ${e.message}")
    }
    sendEvent(name: "connectionStatus", value: "disconnected")
    state.processingDisconnect = false  // Clear disconnect processing flag
}

/**
 * Clears rate limit state to allow manual reconnection
 * Use after rate limit has expired if auto-reconnect hasn't triggered
 */
def clearRateLimit() {
    logInfo("Clearing rate limit state manually")

    // Cancel any scheduled reconnects
    unschedule("connect")

    state.rateLimitedUntil = null
    state.rateLimitedUntilFormatted = null
    state.reconnectAttempts = 0
    state.processingDisconnect = false
    sendEvent(name: "connectionStatus", value: "disconnected")
}

/**
 * Reconnects the stream by disconnecting and reconnecting
 * Useful when events stop flowing - forces Home Connect to send current state
 * Uses runIn() instead of pauseExecution() to prevent double-connect race conditions
 */
def reconnect() {
    logInfo("Reconnecting stream (user-requested)")
    disconnect()
    runIn(2, "connect")
}

/**
 * Shows event statistics for diagnostic purposes
 */
def showEventStats() {
    logInfo("=== Event Statistics ===")
    if (state.eventStats) {
        state.eventStats.each { type, count ->
            logInfo("${type}: ${count} events")
        }
        if (state.lastRealEventTime) {
            def lastRealEventAgo = (now() - state.lastRealEventTime) / 1000
            logInfo("Last real event: ${state.lastRealEventType} (${lastRealEventAgo.toInteger()}s ago)")
        } else {
            logInfo("No real events received yet (only keep-alives)")
        }
    } else {
        logInfo("No events received yet")
    }
    logInfo("======================")
}

/**
 * Clears event statistics
 */
def clearEventStats() {
    logInfo("Clearing event statistics")
    state.eventStats = [:]
    state.lastRealEventTime = null
    state.lastRealEventType = null
    state.applianceLastEventTime = [:]
    state.applianceLastEventType = [:]
    state.applianceLastEventKey = [:]
}

/**
 * Shows detailed stream health diagnostics on demand.
 * Can be run at any time from the device page to see current stream status.
 */
def showStreamHealth() {
    def lastParse = state.lastParseTime ?: 0
    def parseElapsed
    if (lastParse > 0) {
        parseElapsed = now() - lastParse
    } else {
        // lastParseTime missing - use fallback timestamps for diagnostics
        def fallback = state.lastRealEventTime ?: state.connectionStartTime ?: 0
        if (fallback > 0) {
            parseElapsed = now() - fallback
            logWarn("showStreamHealth: lastParseTime is missing, using fallback timestamp")
        } else {
            parseElapsed = 0
        }
    }
    logStreamHealthDiagnostics(parseElapsed)
}

/* ===========================================================================================================
   SSE EVENT HANDLING
   =========================================================================================================== */

/**
 * Called by Hubitat when SSE connection status changes
 * Handles reconnection logic with conservative timing to avoid rate limits
 *
 * Reconnect Strategy:
 * - Normal disconnect (connection was successful): Wait 5 minutes, then reconnect
 * - Failed connection (never connected): Exponential backoff (60s, 120s, 240s, max 300s)
 * - Rate limited: Schedule reconnect for when limit expires + 5 min buffer
 * - Too many follow-up requests: 60 second backoff (same as failed connection)
 *
 * Note: Hubitat may call this method multiple times for a single disconnect event
 * (e.g., both "ERROR" and "STOP" messages). We use state.processingDisconnect to
 * ensure we only schedule one reconnection attempt per disconnect.
 */
def eventStreamStatus(String status) {
    logDebug("Event stream status: ${status}")

    if (status.contains("START")) {
        // Connection successful
        sendEvent(name: "connectionStatus", value: "connected")
        messageBuffer = ""

        state.connectionSucceeded = true
        state.reconnectAttempts = 0
        state.lastConnectTime = now()
        state.connectionStartTime = now()  // Track when connection established
        state.lastParseTime = now()        // Initialize for watchdog
        state.processingDisconnect = false  // Clear disconnect processing flag

        // Ensure persistent watchdog schedule is running
        // Uses cron schedule so it survives hub restarts and state corruption
        unschedule("streamWatchdog")
        schedule("0 */5 * * * ?", "streamWatchdog")

        // Always refresh device status on reconnect to recover any missed events
        def previousDisconnectTime = state.lastDisconnectTime ?: 0
        if (previousDisconnectTime > 0) {
            def disconnectedDuration = now() - previousDisconnectTime
            logInfo("Reconnected after ${(disconnectedDuration/1000).toInteger()}s - refreshing device status")
            runIn(2, "notifyParentReconnected")
        }

    } else if (status.contains("STOP") || status.contains("ERROR")) {
        // Check if we're already processing a disconnect (prevents double-scheduling from ERROR+STOP)
        if (state.processingDisconnect) {
            logDebug("Already processing disconnect - ignoring duplicate event")
            return
        }
        state.processingDisconnect = true

        // Connection lost
        sendEvent(name: "connectionStatus", value: "disconnected")
        state.lastDisconnectTime = now()

        // Check for "Too many follow-up requests" error - needs special handling
        if (status.contains("Too many follow-up requests") || status.contains("ProtocolException")) {
            handleFollowUpRequestsError(status)
            return
        }

        // Don't reconnect if rate limited
        if (state.rateLimitedUntil && now() < state.rateLimitedUntil) {
            logWarn("Rate limited - not reconnecting")
            return
        }

        // Cancel any previously scheduled reconnection attempts
        unschedule("connect")

        // Determine reconnect strategy based on whether we successfully connected
        if (state.connectionSucceeded) {
            // Normal disconnect after successful connection - wait 5 minutes
            // This is the typical idle timeout from Home Connect
            state.connectionSucceeded = false
            logDebug("Normal disconnect - scheduling reconnect in ${NORMAL_RECONNECT_DELAY}s")
            runIn(NORMAL_RECONNECT_DELAY, "connect")
        } else {
            // Connection failed without ever succeeding - use exponential backoff
            def attempts = (state.reconnectAttempts ?: 0) + 1
            state.reconnectAttempts = attempts

            if (attempts > MAX_RECONNECT_ATTEMPTS) {
                logError("Max reconnect attempts (${MAX_RECONNECT_ATTEMPTS}) reached - giving up. Click 'Connect' to retry manually.")
                sendEvent(name: "connectionStatus", value: "failed - manual reconnect required")
                return
            }

            // Exponential backoff: 60s, 120s, 240s, max 300s
            def delay = Math.min(300, 60 * Math.pow(2, attempts - 1) as Integer)
            logWarn("Connection failed - scheduling reconnect in ${delay}s (attempt ${attempts}/${MAX_RECONNECT_ATTEMPTS})")
            runIn(delay, "connect")
        }
    }
}

/**
 * Called by parent app after reconnection to refresh all device status
 */
def notifyParentReconnected() {
    logInfo("Notifying parent to refresh device status")
    parent?.refreshAllDeviceStatus()
}

/**
 * Stream health watchdog - detects silently dead SSE connections
 *
 * Home Connect sends KEEP-ALIVE heartbeats every ~55 seconds. If no data
 * has arrived for STREAM_TIMEOUT (3 minutes), the connection is considered
 * dead and will be automatically reconnected.
 *
 * This handles the case where the TCP connection drops silently without
 * Hubitat detecting it (no STOP/ERROR event from eventStreamStatus).
 */
def streamWatchdog() {
    def currentStatus = device.currentValue("connectionStatus")

    // Handle non-connected states: attempt auto-recovery
    if (currentStatus != "connected") {
        // Check if we're rate limited - don't interfere
        if (state.rateLimitedUntil && now() < state.rateLimitedUntil) {
            logDebug("Stream watchdog: rate limited - waiting for expiry")
            return
        }

        // If we're in a terminal/stuck state, attempt to reconnect
        if (currentStatus == "connecting") {
            // Check for stuck "connecting" state (handled by connect() itself)
            logDebug("Stream watchdog: still connecting - will check again")
            return
        }

        // For all other non-connected states (disconnected, failed, error, etc.),
        // attempt auto-recovery by reconnecting
        logWarn("Stream watchdog: not connected (${currentStatus}) - attempting auto-recovery")
        state.processingDisconnect = false
        state.reconnectAttempts = 0  // Reset attempts so we don't hit the max limit
        connect()
        return
    }

    def lastParse = state.lastParseTime ?: 0
    def elapsed

    if (lastParse > 0) {
        elapsed = now() - lastParse
    } else {
        // lastParseTime is missing/null (possible Hubitat state loss) - use fallback timestamps
        def fallback = state.lastRealEventTime ?: state.connectionStartTime ?: 0
        if (fallback > 0) {
            elapsed = now() - fallback
            logWarn("Stream watchdog: lastParseTime is missing, using fallback timestamp (${(elapsed / 1000).toInteger()}s ago)")
            state.lastParseTime = fallback
        } else {
            logWarn("Stream watchdog: all timestamps missing - forcing reconnect")
            elapsed = STREAM_TIMEOUT + 1
        }
    }

    if (elapsed > STREAM_TIMEOUT) {
        def elapsedSec = (elapsed / 1000).toInteger()
        logWarn("Stream watchdog: no data received for ${elapsedSec}s - stream appears dead, forcing reconnect")
        sendEvent(name: "connectionStatus", value: "disconnected - watchdog timeout")

        // Force close and reconnect
        try {
            interfaces.eventStream.close()
        } catch (Exception e) {
            logDebug("Error closing dead stream: ${e.message}")
        }

        state.connectionSucceeded = false
        state.processingDisconnect = false
        messageBuffer = ""

        // Reconnect after a brief pause
        runIn(5, "connect")
    } else {
        logDebug("Stream watchdog: stream healthy (last data ${(elapsed / 1000).toInteger()}s ago)")

        // Verbose diagnostics: log detailed stream health
        if (settings?.verboseStreamDiag) {
            logStreamHealthDiagnostics(elapsed)
        }
    }
}

/**
 * Logs detailed stream health diagnostics when verbose mode is enabled.
 * Called from streamWatchdog() to help diagnose stale event issues.
 */
private void logStreamHealthDiagnostics(long parseElapsedMs) {
    def lastRealEvent = state.lastRealEventTime ?: 0
    def realEventElapsed = lastRealEvent > 0 ? (now() - lastRealEvent) : -1
    def keepAliveCount = state.eventStats?.get("KEEP-ALIVE") ?: 0
    def connectionAge = state.connectionStartTime ? (now() - state.connectionStartTime) : 0

    logInfo("=== STREAM HEALTH DIAGNOSTICS ===")
    logInfo("Connection age: ${(connectionAge / 1000).toInteger()}s")
    logInfo("Last ANY data (parse): ${(parseElapsedMs / 1000).toInteger()}s ago")
    if (realEventElapsed >= 0) {
        logInfo("Last REAL event: ${(realEventElapsed / 1000).toInteger()}s ago (type: ${state.lastRealEventType ?: 'unknown'})")
    } else {
        logInfo("Last REAL event: NEVER (only KEEP-ALIVEs received)")
    }
    logInfo("KEEP-ALIVE count: ${keepAliveCount}")
    logInfo("Buffer size: ${messageBuffer?.length() ?: 0} chars")

    // Per-appliance event tracking
    def applianceEvents = state.applianceLastEventTime
    if (applianceEvents) {
        logInfo("--- Per-Appliance Last Event ---")
        applianceEvents.each { haId, timestamp ->
            def ago = ((now() - timestamp) / 1000).toInteger()
            def lastType = state.applianceLastEventType?.get(haId) ?: "unknown"
            def lastKey = state.applianceLastEventKey?.get(haId) ?: "unknown"
            logInfo("  ${haId}: ${ago}s ago [${lastType}] ${lastKey}")
        }
    } else {
        logInfo("Per-appliance tracking: no data yet")
    }

    // Event type breakdown
    if (state.eventStats) {
        logInfo("--- Event Type Counts ---")
        state.eventStats.each { type, count ->
            logInfo("  ${type}: ${count}")
        }
    }
    logInfo("=================================")
}

/**
 * Handles "Too many follow-up requests" ProtocolException error
 * This error indicates the HTTP client followed 21+ redirects during the SSE stream
 *
 * Possible causes:
 * - Invalid/expired OAuth token causing redirect loops
 * - API endpoint issues
 * - Network proxy/firewall issues
 *
 * We apply a 60 second backoff and force OAuth token refresh
 */
private void handleFollowUpRequestsError(String status) {
    logWarn("ProtocolException: Too many follow-up requests detected")
    logWarn("This may indicate OAuth token issues or API endpoint problems")

    // Extract the redirect count if available (e.g., "follow-up requests: 21")
    def matcher = status =~ /follow-up requests: (\d+)/
    def redirectCount = 21  // Default
    if (matcher && matcher.size() > 0 && matcher[0].size() > 1) {
        redirectCount = matcher[0][1].toInteger()
    }

    logWarn("HTTP client followed ${redirectCount} redirects before failing")

    // Log connection duration for troubleshooting
    def connectionStart = state.connectionStartTime ?: 0
    if (connectionStart > 0) {
        def duration = (now() - connectionStart) / 1000
        logWarn("Connection lasted ${duration}s before ProtocolException")
    }

    // Log current token validity
    try {
        def tokenExpires = parent?.getTokenExpiryTime() ?: 0
        if (tokenExpires > 0) {
            def tokenAge = (tokenExpires - now()) / 1000
            logWarn("OAuth token still valid for ${tokenAge}s at time of error")
        }
    } catch (Exception e) {
        // Parent app may not have getTokenExpiryTime() yet - ignore
    }

    // Cancel any scheduled reconnects
    unschedule("connect")

    // Force OAuth token refresh on next connection
    parent?.refreshOAuthTokenAndRetry()

    // Use 60 second backoff - same as normal failed connection
    // The duplicate event detection and unschedule() prevents the rapid reconnection loop
    // that was previously occurring. 60s is fast enough for responsive manual device updates.
    def protocolErrorBackoff = 60

    state.connectionSucceeded = false
    state.reconnectAttempts = 0  // Reset attempts since this is a different error class

    logWarn("Scheduling reconnect after ProtocolException: ${protocolErrorBackoff}s")
    sendEvent(name: "connectionStatus", value: "disconnected - protocol error")
    runIn(protocolErrorBackoff, "connect")
}

/**
 * Main entry point for SSE data
 * Called by Hubitat each time data arrives on the event stream
 * 
 * SSE Format:
 * -----------
 * event: STATUS
 * data: {"haId":"...", "items":[...]}
 * 
 * Data may arrive in fragments, so we buffer until we have complete messages
 */
def parse(String text) {
    // Detect driver code updates (e.g. HPM) that don't trigger updated()
    checkDriverVersion()

    // Ignore data if rate limited (prevents processing error responses)
    if (state.rateLimitedUntil && now() < state.rateLimitedUntil) {
        return
    }

    logDebug("Raw SSE data: ${text?.take(200)}${text?.length() > 200 ? '...' : ''}")

    // Track last parse time for stream health watchdog
    state.lastParseTime = now()

    // If we're receiving data, the stream IS connected - correct stale status.
    // Hubitat can fire eventStreamStatus(STOP) while the connection is still alive,
    // leaving connectionStatus as "disconnected" even though parse() is being called.
    def currentStatus = device.currentValue("connectionStatus")
    if (currentStatus != "connected") {
        logInfo("Stream receiving data but status was '${currentStatus}' - correcting to 'connected'")
        sendEvent(name: "connectionStatus", value: "connected")
        state.connectionSucceeded = true
        state.processingDisconnect = false
        // Cancel any pending reconnect since we're actually connected
        unschedule("connect")
    }

    // Update lastEventReceived timestamp on any incoming data
    updateLastEventReceived()

    // Check for rate limit error in the stream
    if (text && (text.contains('"key": "429"') || text.contains('"key":"429"') || text.contains("rate limit"))) {
        handleRateLimitError(text)
        return
    }

    // Buffer incoming data (SSE messages may span multiple parse() calls)
    // Hubitat's eventStream calls parse() once per line with newlines stripped.
    // Re-add line terminators so SSE message boundaries (\n\n) are properly formed.
    // Empty parse() calls represent SSE blank-line delimiters — do NOT discard them.
    messageBuffer += (text ?: "") + "\n"

    // Process complete messages (SSE messages are separated by double newlines)
    while (messageBuffer.contains("\n\n")) {
        def idx = messageBuffer.indexOf("\n\n")
        def message = messageBuffer.substring(0, idx)
        messageBuffer = messageBuffer.substring(idx + 2)
        processSSEMessage(message)
    }

    // Safety: prevent buffer from growing indefinitely if messages lack proper termination
    if (messageBuffer.length() > 10000) {
        logWarn("SSE message buffer exceeded 10KB (${messageBuffer.length()} chars) - clearing stale data")
        messageBuffer = ""
    }
}

/**
 * Updates the lastEventReceived timestamp
 * Called only when actual SSE data arrives to track stream health
 * NOT called on connection attempts or API calls
 */
private void updateLastEventReceived() {
    def now = new Date()
    // ISO 8601 format for programmatic use
    sendEvent(name: "lastEventReceived", value: now.format("yyyy-MM-dd'T'HH:mm:ss.SSSZ"))
    // Human-readable format (legacy attribute)
    sendEvent(name: "lastEventTime", value: now.format("yyyy-MM-dd HH:mm:ss"))
}

/**
 * Updates event statistics to track what types of events we're receiving
 * Helps diagnose issues where stream is connected but not receiving real events
 */
private void updateEventStats(String eventType) {
    if (!state.eventStats) {
        state.eventStats = [:]
    }
    if (!state.eventStats[eventType]) {
        state.eventStats[eventType] = 0
    }
    state.eventStats[eventType] = state.eventStats[eventType] + 1

    // Update last real event time (not keep-alive)
    if (eventType != "KEEP-ALIVE") {
        state.lastRealEventTime = now()
        state.lastRealEventType = eventType
        logDebug("Real event received: ${eventType} (total: ${state.eventStats[eventType]})")
    }
}

/**
 * Tracks per-appliance event timing for diagnostics.
 * Records the last event time, type, and key for each appliance (haId).
 */
private void trackApplianceEvent(String haId, String eventType, String eventKey) {
    if (!state.applianceLastEventTime) state.applianceLastEventTime = [:]
    if (!state.applianceLastEventType) state.applianceLastEventType = [:]
    if (!state.applianceLastEventKey) state.applianceLastEventKey = [:]

    state.applianceLastEventTime[haId] = now()
    state.applianceLastEventType[haId] = eventType
    state.applianceLastEventKey[haId] = eventKey
}

/**
 * Handles rate limit (429) errors from the SSE stream
 * Extracts the retry time and schedules automatic reconnection
 */
private void handleRateLimitError(String text) {
    logError("Rate limit detected in SSE stream - stopping reconnects")
    
    // Extract seconds from error message (e.g., "...remaining period of 86400 seconds")
    def matcher = text =~ /(\d+) seconds/
    def backoffSeconds = 86400  // Default 24 hours if not found
    if (matcher && matcher.size() > 0 && matcher[0].size() > 1) {
        backoffSeconds = matcher[0][1].toInteger()
    }
    
    def rateLimitUntil = now() + (backoffSeconds * 1000)
    state.rateLimitedUntil = rateLimitUntil
    state.reconnectAttempts = 0
    
    def rateLimitTime = formatDateTime(rateLimitUntil)
    state.rateLimitedUntilFormatted = rateLimitTime
    
    sendEvent(name: "connectionStatus", value: "rate limited until ${rateLimitTime}")
    sendEvent(name: "rateLimitRemaining", value: 0)
    logError("Rate limited until ${rateLimitTime}")
    
    // Schedule automatic reconnect when rate limit expires (plus buffer)
    def reconnectDelay = backoffSeconds + RATE_LIMIT_BUFFER
    logInfo("Scheduling automatic reconnect in ${reconnectDelay} seconds (${formatDateTime(now() + reconnectDelay * 1000)})")
    runIn(reconnectDelay, "connect")
}

/**
 * Processes a complete SSE message
 * Extracts event type and data payload, then routes to processEventPayload
 */
private void processSSEMessage(String message) {
    logDebug("Processing SSE message: ${message?.take(100)}${message?.length() > 100 ? '...' : ''}")

    String eventType = null
    String dataPayload = null
    boolean hasDataField = false

    message.split("\n").each { line ->
        if (line.startsWith("event:") && line.length() > 6) {
            eventType = line.substring(6).trim()
        } else if (line.startsWith("data:")) {
            hasDataField = true
            def value = line.length() > 5 ? line.substring(5).trim() : ""
            if (value) dataPayload = value
        }
    }

    // Fallback: handle bare JSON lines (Hubitat may deliver data without "data:" prefix)
    if (!dataPayload) {
        message.split("\n").each { line ->
            if (!dataPayload && line.startsWith("{")) {
                dataPayload = line.trim()
            }
        }
    }

    if (eventType && dataPayload) {
        logDebug("Event type: ${eventType}")
        processEventPayload(dataPayload, eventType)
    } else if (dataPayload) {
        processEventPayload(dataPayload)
    } else if (hasDataField && !dataPayload) {
        // Keep-alive: SSE message with data: field but no payload
        // Home Connect sends these every ~55s without an event: KEEP-ALIVE field
        logDebug("Keep-alive received (empty data field)")
        updateEventStats("KEEP-ALIVE")
    }
}

/**
 * Processes the JSON payload from an SSE event
 * Routes events to the appropriate child device via the parent app
 *
 * Event Types:
 * - KEEP-ALIVE: Connection heartbeat (ignored)
 * - CONNECTED/DISCONNECTED: Appliance online status
 * - STATUS/EVENT/NOTIFY: Appliance state changes (routed to child)
 */
private void processEventPayload(String payload, String eventType = null) {
    if (!payload || !payload.startsWith("{")) return
    
    try {
        def json = new JsonSlurper().parseText(payload)

        String haId = json.haId
        if (!haId) {
            logWarn("Event payload missing haId - ignoring")
            return
        }

        // Handle keep-alive (heartbeat)
        if (eventType == "KEEP-ALIVE") {
            logDebug("Keep-alive received for appliance ${haId}")
            updateEventStats("KEEP-ALIVE")
            if (settings?.verboseStreamDiag) {
                def lastReal = state.applianceLastEventTime?.get(haId)
                def gap = lastReal ? "${((now() - lastReal) / 1000).toInteger()}s" : "never"
                logDebug("KEEP-ALIVE for ${haId} (last real event: ${gap} ago)")
            }
            return
        }

        // Handle appliance connection status
        if (eventType == "DISCONNECTED" || eventType == "CONNECTED") {
            logInfo("Appliance ${haId} is now ${eventType}")
            updateEventStats(eventType)
            trackApplianceEvent(haId, eventType, eventType)
            parent?.handleApplianceConnectionEvent(haId, eventType)
            return
        }

        // Process status/event items and route to child device
        def items = json.items
        if (items instanceof List) {
            items.findAll { it != null && it.key != null }.each { item ->
                def evt = [
                    haId: haId,
                    key: item.key,
                    value: item.value,
                    displayvalue: item.displayvalue ?: item.value?.toString(),
                    unit: item.unit,
                    eventType: eventType
                ]

                logDebug("Received event: ${eventType} - ${item.key} = ${item.value} for ${haId}")
                updateEventStats(eventType ?: "STATUS")
                trackApplianceEvent(haId, eventType ?: "STATUS", item.key)
                parent?.handleApplianceEvent(evt)
            }
        } else {
            logDebug("Event payload has no items array - eventType: ${eventType}, haId: ${haId}")
        }

    } catch (Exception e) {
        logError("Error parsing event payload: ${e.message}")
        logError("Payload: ${payload?.take(200)}")
    }
}

/* ===========================================================================================================
   HOME CONNECT API - CORE HTTP METHODS
   =========================================================================================================== */

/**
 * Performs an HTTP GET request to the Home Connect API
 * Includes automatic token refresh on 401 errors
 *
 * @param path API endpoint path (e.g., "/api/homeappliances/{haId}/status")
 * @param closure Callback to receive the response data
 */
def apiGet(String path, Closure closure) {
    def token = parent?.getOAuthToken()
    def language = parent?.getLanguage() ?: "en-US"
    
    if (!token) {
        logError("No OAuth token for API GET")
        return
    }
    
    logDebug("API GET: ${path}")
    trackApiCall("GET", path)

    try {
        httpGet(
            uri: getApiUrl() + path,
            contentType: "application/json",
            headers: [
                'Authorization': "Bearer ${token}",
                'Accept-Language': language,
                'Accept': "application/vnd.bsh.sdk.v1+json"
            ]
        ) { response ->
            logDebug("API GET response status: ${response.status}")
            
            // Extract rate limit headers
            extractRateLimitHeaders(response)
            
            if (response.data) {
                closure(response.data)
            }
        }
    } catch (groovyx.net.http.HttpResponseException e) {
        handleHttpError("GET", path, e, closure)
    } catch (Exception e) {
        logError("API GET error: ${e.message} - path: ${path}")
    }
}

/**
 * Performs an HTTP PUT request to the Home Connect API
 * Used for setting values (power state, programs, options)
 *
 * @param path API endpoint path
 * @param data Map of data to send in request body
 * @param closure Callback to receive the response data
 */
def apiPut(String path, Map data, Closure closure) {
    def token = parent?.getOAuthToken()
    def language = parent?.getLanguage() ?: "en-US"
    
    if (!token) {
        logError("No OAuth token for API PUT")
        return
    }
    
    // Check rate limit before making request
    if (state.rateLimitedUntil && now() < state.rateLimitedUntil) {
        logWarn("API PUT blocked - rate limited")
        return
    }
    
    String body = new JsonOutput().toJson(data)
    logDebug("API PUT: ${path}")
    trackApiCall("PUT", path)

    try {
        httpPut(
            uri: getApiUrl() + path,
            contentType: "application/json",
            requestContentType: "application/json",
            body: body,
            headers: [
                'Authorization': "Bearer ${token}",
                'Accept-Language': language,
                'Accept': "application/vnd.bsh.sdk.v1+json"
            ]
        ) { response ->
            logDebug("API PUT response status: ${response.status}")
            
            // Extract rate limit headers
            extractRateLimitHeaders(response)
            
            // Always call closure - successful PUT may return 204 No Content with empty body
            closure(response.data ?: [status: response.status])
        }
    } catch (groovyx.net.http.HttpResponseException e) {
        handleHttpError("PUT", path, e, closure)
    } catch (Exception e) {
        logError("API PUT error: ${e.message} - path: ${path}")
    }
}

/**
 * Performs an HTTP DELETE request to the Home Connect API
 * Used for stopping programs
 *
 * @param path API endpoint path
 * @param closure Callback to receive the response data
 */
def apiDelete(String path, Closure closure) {
    def token = parent?.getOAuthToken()
    def language = parent?.getLanguage() ?: "en-US"
    
    if (!token) {
        logError("No OAuth token for API DELETE")
        return
    }
    
    // Check rate limit before making request
    if (state.rateLimitedUntil && now() < state.rateLimitedUntil) {
        logWarn("API DELETE blocked - rate limited")
        return
    }
    
    logDebug("API DELETE: ${path}")
    trackApiCall("DELETE", path)

    try {
        httpDelete(
            uri: getApiUrl() + path,
            contentType: "application/json",
            requestContentType: "application/json",
            headers: [
                'Authorization': "Bearer ${token}",
                'Accept-Language': language,
                'Accept': "application/vnd.bsh.sdk.v1+json"
            ]
        ) { response ->
            logDebug("API DELETE response status: ${response.status}")
            
            // Extract rate limit headers
            extractRateLimitHeaders(response)
            
            // Always call closure - successful DELETE may return 204 No Content with empty body
            closure(response.data ?: [status: response.status])
        }
    } catch (groovyx.net.http.HttpResponseException e) {
        handleHttpError("DELETE", path, e, closure)
    } catch (Exception e) {
        logError("API DELETE error: ${e.message} - path: ${path}")
    }
}

/**
 * Extracts rate limit information from API response headers
 * Updates rateLimitRemaining and rateLimitLimit attributes
 *
 * Home Connect returns these headers on API responses:
 * - X-RateLimit-Limit: Total allowed calls (typically 1000)
 * - X-RateLimit-Remaining: Calls remaining in the current period
 */
private void extractRateLimitHeaders(response) {
    try {
        def headers = response.headers
        if (headers == null) {
            logDebug("extractRateLimitHeaders: response.headers is null")
            return
        }

        // Log all header names at debug level to diagnose missing rate-limit headers
        def headerNames = []
        headers.each { headerNames << it.name }
        logDebug("Response headers: ${headerNames}")

        // Hubitat httpGet response headers are Apache Header objects; iterate to find rate-limit headers
        String remaining = null
        String limit = null
        headers.each { hdr ->
            if (hdr.name?.equalsIgnoreCase('X-RateLimit-Remaining')) remaining = hdr.value
            if (hdr.name?.equalsIgnoreCase('X-RateLimit-Limit'))      limit     = hdr.value
            // Also check without the X- prefix in case the API varies
            if (hdr.name?.equalsIgnoreCase('RateLimit-Remaining'))    remaining = hdr.value
            if (hdr.name?.equalsIgnoreCase('RateLimit-Limit'))        limit     = hdr.value
        }

        logDebug("Rate limit header values - remaining: ${remaining}, limit: ${limit}")

        if (remaining != null) {
            def remainingInt = remaining.toString().trim().toInteger()
            def previousRemaining = device.currentValue("rateLimitRemaining") as Integer
            sendEvent(name: "rateLimitRemaining", value: remainingInt)
            logDebug("Rate limit remaining: ${remainingInt}")

            // Detect Home Connect window reset: remaining jumped up means their window rolled over
            if (previousRemaining != null && remainingInt > previousRemaining) {
                logInfo("Home Connect rate limit window reset detected (${previousRemaining} -> ${remainingInt}), resetting local counter")
                state.apiCallsTotal = 0
                state.apiCallsByMethod = [GET: 0, PUT: 0, DELETE: 0]
                state.apiCallsByCategory = [:]
                sendEvent(name: "apiCallsToday", value: 0)
            }

            // Warn if getting low
            if (remainingInt < 100) {
                logWarn("Rate limit warning: only ${remainingInt} API calls remaining")
            }
        }

        if (limit != null) {
            def limitInt = limit.toString().trim().toInteger()
            sendEvent(name: "rateLimitLimit", value: limitInt)
            logDebug("Rate limit total: ${limitInt}")
        }
    } catch (Exception e) {
        logWarn("Could not extract rate limit headers: ${e.message}")
    }
}

/**
 * Tracks each outgoing API call. Counter resets are driven by the API's own
 * X-RateLimit-Remaining header (detected in extractRateLimitHeaders) rather
 * than a guessed 24-hour timer, so our window always matches Home Connect's.
 */
private void trackApiCall(String method, String path) {
    try {
        // Initialise state on first call
        if (state.apiCallsTotal == null) state.apiCallsTotal = 0
        if (state.apiCallsByMethod == null) state.apiCallsByMethod = [GET: 0, PUT: 0, DELETE: 0]
        if (state.apiCallsByCategory == null) state.apiCallsByCategory = [:]

        state.apiCallsTotal = state.apiCallsTotal + 1
        state.apiCallsByMethod[method] = (state.apiCallsByMethod[method] ?: 0) + 1

        // Categorise by endpoint path
        def category = categorizeApiCall(path)
        state.apiCallsByCategory[category] = (state.apiCallsByCategory[category] ?: 0) + 1

        sendEvent(name: "apiCallsToday", value: state.apiCallsTotal)

        // Warnings based on API-reported remaining (authoritative) if available,
        // otherwise fall back to our own counter
        def remaining = device.currentValue("rateLimitRemaining")
        if (remaining != null) {
            def remainingInt = remaining as Integer
            if (remainingInt == 200 || remainingInt == 100 || remainingInt == 50) {
                logWarn("Rate limit warning: ${remainingInt} API calls remaining (${state.apiCallsTotal} calls tracked this window)")
            } else if (remainingInt <= 20 && remainingInt % 5 == 0) {
                logError("Rate limit critical: only ${remainingInt} API calls remaining!")
            }
        }
    } catch (Exception e) {
        logDebug("Error tracking API call: ${e.message}")
    }
}

/**
 * Maps an API path to a human-readable category for usage reporting.
 */
private String categorizeApiCall(String path) {
    if (path == null) return "unknown"
    if (path.contains("/programs/available")) return "programs"
    if (path.contains("/programs/active"))    return "programs"
    if (path.contains("/programs/selected"))  return "programs"
    if (path.contains("/status"))             return "status"
    if (path.contains("/settings"))           return "settings"
    if (path.contains("/commands"))           return "commands"
    if (path.contains("/images"))             return "images"
    if (path.endsWith("/homeappliances"))     return "discovery"
    return "other"
}

/**
 * Command: show API call usage for the current 24-hour window.
 */
def showApiUsage() {
    def total = state.apiCallsTotal ?: 0
    def byMethod = state.apiCallsByMethod ?: [:]
    def byCategory = state.apiCallsByCategory ?: [:]
    def remaining = device.currentValue("rateLimitRemaining")
    def limit = device.currentValue("rateLimitLimit")

    log.info "==================== API USAGE ===================="
    log.info "Calls tracked this window: ${total}"
    if (remaining != null && limit != null) {
        log.info "API-reported: ${remaining} / ${limit} remaining"
    } else if (remaining != null) {
        log.info "API-reported remaining: ${remaining}"
    }
    log.info "By method:   GET=${byMethod.GET ?: 0}  PUT=${byMethod.PUT ?: 0}  DELETE=${byMethod.DELETE ?: 0}"
    byCategory.sort().each { cat, count ->
        log.info "  ${cat}: ${count}"
    }
    log.info "(Counter resets automatically when Home Connect's window resets)"
    log.info "==================================================="
}

/**
 * Command: reset the API call counter (e.g. after manual verification).
 */
def resetApiCounter() {
    state.apiCallsTotal = 0
    state.apiCallsByMethod = [GET: 0, PUT: 0, DELETE: 0]
    state.apiCallsByCategory = [:]
    sendEvent(name: "apiCallsToday", value: 0)
    logInfo("API call counter reset")
}

/**
 * Handles HTTP errors from API calls
 * Implements retry logic for 401 (token expired) errors
 */
private void handleHttpError(String method, String path, groovyx.net.http.HttpResponseException e, Closure closure) {
    def statusCode = e.getStatusCode()
    def responseData = e.getResponse()?.getData()
    
    // Try to extract rate limit headers even from error responses
    try {
        extractRateLimitHeaders(e.getResponse())
    } catch (Exception ex) {
        // Ignore - not all error responses have headers accessible
    }
    
    switch (statusCode) {
        case 401:
            logWarn("API ${method} 401 Unauthorized - refreshing token")
            if (parent?.refreshOAuthTokenAndRetry()) {
                logInfo("Retrying API ${method} after token refresh")
                // Retry based on method type
                switch (method) {
                    case "GET": apiGetRetry(path, closure); break
                    case "PUT": logWarn("PUT retry not implemented - please retry manually"); break
                    case "DELETE": apiDeleteRetry(path, closure); break
                }
            }
            break
            
        case 404:
            if (path.contains('programs/active')) {
                // No active program - expected when appliance is idle
                // Don't throw exception as it gets logged as ERROR by Hubitat
                // The closure simply won't be called, which the parent app handles gracefully
                logDebug("No active program (appliance idle)")
            } else {
                logWarn("API ${method} 404 Not Found: ${path}")
            }
            break
            
        case 409:
            // 409 Conflict - command rejected by appliance
            def reasonInfo = extractConflictReason(responseData, path)
            logDebug("API ${method} 409 raw response: ${responseData}")
            logDebug("API ${method} 409 path: ${path}")

            if (method == "GET") {
                // GET 409 occurs during status initialization - not a command failure
                logInfo("API GET 409 - ${reasonInfo.message}")
            } else {
                // PUT/DELETE 409 is a command rejection - always log visibly
                logWarn("API ${method} 409 - ${reasonInfo.message}")
                def haId = extractHaIdFromPath(path)
                if (haId) {
                    parent?.handleCommandError(haId, reasonInfo.errorType, reasonInfo.message)
                }
            }
            break
            
        case 429:
            logError("API ${method} 429 Rate Limited")
            sendEvent(name: "rateLimitRemaining", value: 0)
            state.rateLimitedUntil = now() + 60000  // Back off for 1 minute
            break
            
        case 503:
            logWarn("API ${method} 503 Service Unavailable - appliance may be offline")

            // Extract haId from path to notify the device
            def haId503 = extractHaIdFromPath(path)
            if (haId503) {
                parent?.handleCommandError(haId503, "Service unavailable", "appliance may be offline or network issue")
            }
            break
            
        default:
            logError("API ${method} error ${statusCode}: ${responseData} - path: ${path}")
    }
}

/**
 * Extracts a user-friendly conflict reason from 409 error response
 * Returns a map with 'message' (user-friendly) and 'errorType' (category)
 */
private Map extractConflictReason(def responseData, String path = null) {
    def target = extractTargetFromPath(path)

    if (!responseData) {
        return [errorType: "Conflict", message: "Command rejected - appliance not ready (check door, power, remote control)"]
    }

    // Try to extract the error key and description from the API response
    try {
        def errorKey = responseData?.error?.key ?: ""
        def errorDesc = responseData?.error?.description ?: ""
        if (errorKey) {
            return translateApiError(errorKey, errorDesc, target)
        }
    } catch (Exception e) {
        // Fall through to keyword matching
    }

    // Fallback: keyword matching on raw response
    def errorString = responseData.toString().toLowerCase()
    if (errorString.contains("door")) {
        return [errorType: "Door", message: "Command rejected - door may be open${target ? ' (' + target + ')' : ''}"]
    } else if (errorString.contains("remote")) {
        return [errorType: "Remote Control", message: "Command rejected - remote control not active or not allowed"]
    } else if (errorString.contains("power")) {
        return [errorType: "Power", message: "Command rejected - appliance may be off or in standby"]
    } else if (errorString.contains("operation")) {
        return [errorType: "Operation", message: "Command rejected - appliance is busy or transitioning states"]
    } else {
        return [errorType: "Conflict", message: "Command rejected - appliance not ready (check door, power, remote control)"]
    }
}

/**
 * Translates Home Connect API error keys into user-friendly messages
 */
private Map translateApiError(String errorKey, String errorDesc, String target) {
    def shortKey = errorKey.contains(".") ? errorKey.substring(errorKey.lastIndexOf(".") + 1) : errorKey
    def targetInfo = target ? " '${target}'" : ""

    switch (shortKey) {
        case "UnsupportedSetting":
            return [errorType: "Not Supported", message: "Setting${targetInfo} is not supported by this appliance"]
        case "UnsupportedOption":
            return [errorType: "Not Supported", message: "Option${targetInfo} is not supported by this appliance"]
        case "UnsupportedCommand":
            return [errorType: "Not Supported", message: "Command${targetInfo} is not supported by this appliance"]
        case "UnsupportedOperation":
            return [errorType: "Not Supported", message: "Operation${targetInfo} is not supported by this appliance in its current state"]
        case "UnsupportedProgram":
            return [errorType: "Not Supported", message: "Program${targetInfo} is not supported by this appliance"]
        case "InvalidOptionValue":
            return [errorType: "Invalid Value", message: "Invalid value for${targetInfo}${errorDesc ? ' - ' + errorDesc : ''}"]
        case "InvalidSettingValue":
            return [errorType: "Invalid Value", message: "Invalid value for setting${targetInfo}${errorDesc ? ' - ' + errorDesc : ''}"]
        case "ReadOnlySetting":
            return [errorType: "Read Only", message: "Setting${targetInfo} is read-only and cannot be changed"]
        case "DoorOpen":
            return [errorType: "Door", message: "Command rejected - door is open"]
        case "NotInRemoteControlMode":
        case "RemoteControlNotActive":
            return [errorType: "Remote Control", message: "Remote control is not active on the appliance - enable it on the appliance first"]
        case "RemoteStartNotAllowed":
        case "NotInRemoteStartMode":
            return [errorType: "Remote Start", message: "Remote start is not allowed - enable remote start on the appliance first"]
        case "ActiveProgramNotSet":
            return [errorType: "No Program", message: "No program is currently active"]
        case "ProgramNotAvailable":
            return [errorType: "Not Available", message: "Program${targetInfo} is not available in the current state"]
        case "InvalidSettingState":
            return [errorType: "Not Available", message: "Setting${targetInfo} is currently not available or writable"]
        case "SelectedProgramNotSet":
            return [errorType: "No Program", message: "No program is selected - select or start a program first before setting options"]
        case "AlreadyInSelectedState":
            return [errorType: "Already Set", message: "Appliance is already in the requested state"]
        case "WrongOperationState":
            return [errorType: "Wrong State", message: "Command rejected - appliance is not in the right state${errorDesc ? ' (' + errorDesc + ')' : ''}"]
        default:
            // For unknown error keys, show the key and description
            return [errorType: shortKey, message: "${errorKey}${errorDesc ? ': ' + errorDesc : ''}${targetInfo ? ' for' + targetInfo : ''}"]
    }
}

/**
 * Extracts the setting/program name from an API path for context in error messages
 * e.g., /api/homeappliances/xxx/settings/Cooking.Common.Setting.Lighting → Lighting
 * e.g., /api/homeappliances/xxx/programs/active → active program
 */
private String extractTargetFromPath(String path) {
    if (!path) return null

    try {
        // Settings path: .../settings/Some.Setting.Key
        if (path.contains("/settings/")) {
            def settingKey = path.substring(path.lastIndexOf("/settings/") + 10)
            if (settingKey && !settingKey.isEmpty()) {
                // Extract the last part of the dotted key for readability
                return settingKey.contains(".") ? settingKey.substring(settingKey.lastIndexOf(".") + 1) : settingKey
            }
        }
        // Programs path: .../programs/active
        if (path.contains("/programs/active")) {
            return null  // Program name is in the request body, not the path
        }
        // Options path: .../options/Some.Option.Key
        if (path.contains("/options/")) {
            def optionKey = path.substring(path.lastIndexOf("/options/") + 9)
            if (optionKey && !optionKey.isEmpty()) {
                return optionKey.contains(".") ? optionKey.substring(optionKey.lastIndexOf(".") + 1) : optionKey
            }
        }
    } catch (Exception e) {
        // Don't fail on path parsing
    }
    return null
}

/**
 * Extracts haId from API path
 * Example: /api/homeappliances/SIEMENS-HCS02DWH1-6C6FA08A26F1/programs/active -> SIEMENS-HCS02DWH1-6C6FA08A26F1
 */
private String extractHaIdFromPath(String path) {
    if (!path) return null

    def matcher = path =~ /\/api\/homeappliances\/([^\/]+)/
    if (matcher && matcher.size() > 0 && matcher[0].size() > 1) {
        return matcher[0][1]
    }
    return null
}

/**
 * Retry helper for GET requests after token refresh
 */
private void apiGetRetry(String path, Closure closure) {
    def token = parent?.getOAuthToken()
    def language = parent?.getLanguage() ?: "en-US"

    if (!token) {
        logError("No OAuth token for API GET retry")
        return
    }

    trackApiCall("GET", path)

    try {
        httpGet(
            uri: getApiUrl() + path,
            contentType: "application/json",
            headers: [
                'Authorization': "Bearer ${token}",
                'Accept-Language': language,
                'Accept': "application/vnd.bsh.sdk.v1+json"
            ]
        ) { response ->
            logDebug("API GET retry response: ${response.status}")
            extractRateLimitHeaders(response)
            if (response.data) {
                closure(response.data)
            }
        }
    } catch (Exception e) {
        logError("API GET retry failed: ${e.message}")
    }
}

/**
 * Retry helper for DELETE requests after token refresh
 */
private void apiDeleteRetry(String path, Closure closure) {
    def token = parent?.getOAuthToken()
    def language = parent?.getLanguage() ?: "en-US"

    if (!token) {
        logError("No OAuth token for API DELETE retry")
        return
    }

    trackApiCall("DELETE", path)

    try {
        httpDelete(
            uri: getApiUrl() + path,
            contentType: "application/json",
            requestContentType: "application/json",
            headers: [
                'Authorization': "Bearer ${token}",
                'Accept-Language': language,
                'Accept': "application/vnd.bsh.sdk.v1+json"
            ]
        ) { response ->
            logDebug("API DELETE retry response: ${response.status}")
            extractRateLimitHeaders(response)
            if (response.data) {
                closure(response.data)
            }
        }
    } catch (Exception e) {
        logError("API DELETE retry failed: ${e.message}")
    }
}

/* ===========================================================================================================
   HOME CONNECT API - APPLIANCE METHODS
   =========================================================================================================== */

/**
 * Retrieves list of all Home Connect appliances registered to the user
 */
def getHomeAppliances(Closure closure) {
    logDebug("Retrieving all Home Appliances")
    apiGet("${ENDPOINT_APPLIANCES}") { response ->
        closure(response.data?.homeappliances ?: [])
    }
}

/**
 * Retrieves details for a specific appliance
 */
def getHomeAppliance(String haId, Closure closure) {
    logDebug("Retrieving appliance ${haId}")
    apiGet("${ENDPOINT_APPLIANCES}/${haId}") { response ->
        closure(response.data)
    }
}

/* ===========================================================================================================
   HOME CONNECT API - PROGRAM METHODS
   =========================================================================================================== */

/**
 * Gets list of available programs for an appliance
 */
def getAvailablePrograms(String haId, Closure closure) {
    logDebug("Retrieving available programs for ${haId}")
    apiGet("${ENDPOINT_APPLIANCES}/${haId}/programs/available") { response ->
        closure(response.data?.programs ?: [])
    }
}

/**
 * Gets details for a specific available program (including options)
 */
def getAvailableProgram(String haId, String programKey, Closure closure) {
    logDebug("Retrieving program ${programKey} for ${haId}")
    apiGet("${ENDPOINT_APPLIANCES}/${haId}/programs/available/${programKey}") { response ->
        closure(response.data)
    }
}

/**
 * Gets the currently active (running) program
 */
def getActiveProgram(String haId, Closure closure) {
    logDebug("Retrieving active program for ${haId}")
    apiGet("${ENDPOINT_APPLIANCES}/${haId}/programs/active") { response ->
        closure(response.data)
    }
}

/**
 * Starts a program on the appliance
 * 
 * @param haId Appliance ID
 * @param programKey Program key (e.g., "Dishcare.Dishwasher.Program.Eco50")
 * @param options Optional program options
 */
def setActiveProgram(String haId, String programKey, def options = "", Closure closure) {
    def data = [key: programKey]
    if (options != "") {
        data.put("options", options)
    }
    logInfo("Starting program ${programKey} on ${haId}")
    apiPut("${ENDPOINT_APPLIANCES}/${haId}/programs/active", [data: data]) { response ->
        closure(response.data)
    }
}

/**
 * Stops the currently running program
 */
def stopActiveProgram(String haId, Closure closure) {
    logInfo("Stopping program on ${haId}")
    apiDelete("${ENDPOINT_APPLIANCES}/${haId}/programs/active") { response ->
        closure(response.data)
    }
}

/**
 * Gets the currently selected (but not started) program
 */
def getSelectedProgram(String haId, Closure closure) {
    logDebug("Retrieving selected program for ${haId}")
    apiGet("${ENDPOINT_APPLIANCES}/${haId}/programs/selected") { response ->
        closure(response.data)
    }
}

/**
 * Sets the selected program (without starting it)
 */
def setSelectedProgram(String haId, String programKey, def options = "", Closure closure) {
    def data = [key: programKey]
    if (options != "") {
        data.put("options", options)
    }
    logDebug("Setting selected program ${programKey} on ${haId}")
    apiPut("${ENDPOINT_APPLIANCES}/${haId}/programs/selected", [data: data]) { response ->
        closure(response.data)
    }
}

/**
 * Sets an option on the selected program
 */
def setSelectedProgramOption(String haId, String optionKey, def optionValue, Closure closure) {
    def data = [key: optionKey, value: optionValue]
    logDebug("Setting program option ${optionKey}=${optionValue} on ${haId}")
    apiPut("${ENDPOINT_APPLIANCES}/${haId}/programs/selected/options/${optionKey}", [data: data]) { response ->
        closure(response.data)
    }
}

/* ===========================================================================================================
   HOME CONNECT API - STATUS & SETTINGS METHODS
   =========================================================================================================== */

/**
 * Gets current status of an appliance (operation state, door state, etc.)
 */
def getStatus(String haId, Closure closure) {
    logDebug("Retrieving status for ${haId}")
    apiGet("${ENDPOINT_APPLIANCES}/${haId}/status") { response ->
        closure(response.data?.status ?: [])
    }
}

/**
 * Gets current settings of an appliance (power state, etc.)
 */
def getSettings(String haId, Closure closure) {
    logDebug("Retrieving settings for ${haId}")
    apiGet("${ENDPOINT_APPLIANCES}/${haId}/settings") { response ->
        closure(response.data?.settings ?: [])
    }
}

/**
 * Sets a setting on an appliance (e.g., power state)
 */
def setSetting(String haId, String settingKey, def value, Closure closure) {
    logInfo("Setting ${settingKey}=${value} on ${haId}")
    apiPut("${ENDPOINT_APPLIANCES}/${haId}/settings/${settingKey}", [data: [key: settingKey, value: value]]) { response ->
        closure(response.data)
    }
}

/* ===========================================================================================================
   UTILITY METHODS
   =========================================================================================================== */

/**
 * Converts a map to a URL query string
 */
def toQueryString(Map m) {
    return m.collect { k, v -> "${k}=${new URI(null, null, v.toString(), null)}" }.sort().join("&")
}

/**
 * Converts seconds to HH:MM format
 */
def convertSecondsToTime(Integer sec) {
    if (!sec || sec <= 0) return "00:00"
    long hours = java.util.concurrent.TimeUnit.SECONDS.toHours(sec)
    long minutes = java.util.concurrent.TimeUnit.SECONDS.toMinutes(sec) % 60
    return String.format("%02d:%02d", hours, minutes)
}

/**
 * Extracts the last segment from a dotted enum value
 * e.g., "BSH.Common.EnumType.PowerState.On" → "On"
 */
def extractEnumValue(String full) {
    if (!full) return null
    return full.substring(full.lastIndexOf(".") + 1)
}

/**
 * Formats a timestamp as human-readable date/time
 */
private String formatDateTime(Long timestamp) {
    def date = new Date(timestamp)
    return date.format("yyyy-MM-dd h:mm a", location?.timeZone ?: TimeZone.getDefault())
}

/**
 * Returns map of supported Home Connect languages/regions
 */
def getSupportedLanguages() {
    return [
        "Bulgarian": ["Bulgaria": "bg-BG"],
        "Chinese (Simplified)": ["China": "zh-CN", "Hong Kong": "zh-HK", "Taiwan": "zh-TW"],
        "Czech": ["Czech Republic": "cs-CZ"],
        "Danish": ["Denmark": "da-DK"],
        "Dutch": ["Belgium": "nl-BE", "Netherlands": "nl-NL"],
        "English": ["Australia": "en-AU", "Canada": "en-CA", "India": "en-IN", "New Zealand": "en-NZ", 
                    "Singapore": "en-SG", "South Africa": "en-ZA", "United Kingdom": "en-GB", "United States": "en-US"],
        "Finnish": ["Finland": "fi-FI"],
        "French": ["Belgium": "fr-BE", "Canada": "fr-CA", "France": "fr-FR", "Luxembourg": "fr-LU", "Switzerland": "fr-CH"],
        "German": ["Austria": "de-AT", "Germany": "de-DE", "Luxembourg": "de-LU", "Switzerland": "de-CH"],
        "Greek": ["Greece": "el-GR"],
        "Hungarian": ["Hungary": "hu-HU"],
        "Italian": ["Italy": "it-IT", "Switzerland": "it-CH"],
        "Norwegian": ["Norway": "nb-NO"],
        "Polish": ["Poland": "pl-PL"],
        "Portuguese": ["Portugal": "pt-PT"],
        "Romanian": ["Romania": "ro-RO"],
        "Russian": ["Russian Federation": "ru-RU"],
        "Serbian": ["Serbia": "sr-SR"],
        "Slovak": ["Slovakia": "sk-SK"],
        "Slovenian": ["Slovenia": "sl-SI"],
        "Spanish": ["Chile": "es-CL", "Peru": "es-PE", "Spain": "es-ES"],
        "Swedish": ["Sweden": "sv-SE"],
        "Turkish": ["Turkey": "tr-TR"],
        "Ukrainian": ["Ukraine": "uk-UA"]
    ]
}

/**
 * Flattens the language map for use in preference dropdowns
 */
def toFlattenedLanguageMap(Map m) {
    return m.collectEntries { k, v ->
        def flattened = [:]
        if (v instanceof Map) {
            v.each { k1, v1 ->
                flattened << ["${v1}": "${k} - ${k1} (${v1})"]
            }
        } else {
            flattened << ["${k}": v]
        }
        return flattened
    }
}

/* ===========================================================================================================
   LOGGING METHODS
   =========================================================================================================== */

/**
 * Internal command for logging (z_ prefix convention)
 * Can be called from other components if needed
 */
def z_deviceLog(String level, String msg) {
    switch (level) {
        case "debug": logDebug(msg); break
        case "info": logInfo(msg); break
        case "warn": logWarn(msg); break
        case "error": logError(msg); break
        default: log.info "Home Connect Stream: ${msg}"
    }
}

/**
 * Sets the API URL to use for Home Connect calls
 * Called by parent app - allows different apps to use different endpoints (production vs simulator)
 */
def z_setApiUrl(String url) {
    state.apiUrl = url
    sendEvent(name: "apiUrl", value: url)
    logInfo("API URL set to: ${url}")
}

private void logDebug(String msg) {
    if (debugLogging) {
        log.debug "Home Connect Stream: ${msg}"
    }
}

private void logInfo(String msg) {
    log.info "Home Connect Stream: ${msg}"
}

private void logWarn(String msg) {
    log.warn "Home Connect Stream: ${msg}"
}

private void logError(String msg) {
    log.error "Home Connect Stream: ${msg}"
}