import * as fs from 'fs-extra';
import * as path from 'path';
import express from 'express';
import { Router } from 'express';
import { ParquetWriter } from './parquet-writer';
import {
  SignalKPlugin,
  PluginConfig,
  PathConfig,
  DataRecord,
  PluginState,
  TypedRequest,
  TypedResponse,
  PathsApiResponse,
  FilesApiResponse,
  QueryApiResponse,
  SampleApiResponse,
  ConfigApiResponse,
  HealthApiResponse,
  S3TestApiResponse,
  QueryRequest,
  PathConfigRequest,
  PathInfo,
  WebAppPathConfig,
  CommandConfig,
  CommandRegistrationState,
  CommandExecutionRequest,
  CommandRegistrationRequest,
  CommandApiResponse,
  CommandExecutionResponse,
  CommandPutHandler,
  CommandExecutionResult,
  CommandHistoryEntry,
  NormalizedDelta,
} from './types';
import { glob } from 'glob';
import {
  Context,
  Delta,
  hasValues,
  Path,
  PathValue,
  ServerAPI,
  SourceRef,
  Update,
  Timestamp,
} from '@signalk/server-api';

// AWS S3 for file upload
// eslint-disable-next-line @typescript-eslint/no-explicit-any
let S3Client: any,
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  PutObjectCommand: any,
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  ListObjectsV2Command: any,
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  HeadObjectCommand: any;
import('@aws-sdk/client-s3')
  .then(s3 => {
    S3Client = s3.S3Client;
    PutObjectCommand = s3.PutObjectCommand;
    ListObjectsV2Command = s3.ListObjectsV2Command;
    HeadObjectCommand = s3.HeadObjectCommand;
  })
  .catch(() => {
    // eslint-disable-next-line no-console
    console.warn('AWS S3 SDK not available for file uploads');
  });

// DuckDB for webapp queries
import { DuckDBInstance } from '@duckdb/node-api';
import { registerHistoryApiRoute } from './HistoryAPI';

// Global variables for path and command management
let currentPaths: PathConfig[] = [];
let currentCommands: CommandConfig[] = [];
let commandHistory: CommandHistoryEntry[] = [];
let appInstance: ServerAPI;

// Command state management
const commandState: CommandRegistrationState = {
  registeredCommands: new Map<string, CommandConfig>(),
  putHandlers: new Map<string, CommandPutHandler>(),
};

// Enhanced function to load webapp configuration with backward compatibility
function loadWebAppConfig(): WebAppPathConfig {
  const webAppConfigPath = path.join(
    appInstance.getDataDirPath(),
    'signalk-parquet',
    'webapp-config.json'
  );

  try {
    if (fs.existsSync(webAppConfigPath)) {
      const configData = fs.readFileSync(webAppConfigPath, 'utf8');
      const rawConfig = JSON.parse(configData);

      // Migrate old config format to new format with backward compatibility
      const migratedPaths = (rawConfig.paths || []).map(
        (path: Partial<PathConfig>) => ({
          path: path.path,
          name: path.name,
          enabled: path.enabled,
          regimen: path.regimen,
          source: path.source || undefined, // Add missing source field for streamer branch
          context: path.context,
          excludeMMSI: path.excludeMMSI,
        })
      );

      const migratedCommands = rawConfig.commands || [];

      const migratedConfig = {
        paths: migratedPaths,
        commands: migratedCommands,
      };

      appInstance.debug(
        `Loaded and migrated ${migratedPaths.length} paths and ${migratedCommands.length} commands from existing config`
      );

      // Save the migrated config back to preserve the new format
      try {
        fs.writeFileSync(
          webAppConfigPath,
          JSON.stringify(migratedConfig, null, 2)
        );
        appInstance.debug(
          'Saved migrated configuration with source field compatibility'
        );
      } catch (saveError) {
        appInstance.debug(
          `Warning: Could not save migrated config: ${saveError}`
        );
      }

      return migratedConfig;
    }
  } catch (error) {
    appInstance.error(`Failed to load webapp configuration: ${error}`);

    // BACKUP the broken file instead of destroying it
    try {
      const backupPath = webAppConfigPath + '.backup.' + Date.now();
      if (fs.existsSync(webAppConfigPath)) {
        fs.copyFileSync(webAppConfigPath, backupPath);
        appInstance.debug(`Backed up broken config to: ${backupPath}`);
      }
    } catch (backupError) {
      appInstance.debug(`Could not backup broken config: ${backupError}`);
    }
  }

  // Only create defaults if NO config file exists
  if (!fs.existsSync(webAppConfigPath)) {
    const defaultConfig = getDefaultWebAppConfig();
    appInstance.debug(
      'No existing configuration found, using default installation'
    );

    // Save the default configuration for future use
    try {
      const configDir = path.dirname(webAppConfigPath);
      if (!fs.existsSync(configDir)) {
        fs.mkdirSync(configDir, { recursive: true });
      }
      fs.writeFileSync(
        webAppConfigPath,
        JSON.stringify(defaultConfig, null, 2)
      );
      appInstance.debug('Saved default installation configuration');
    } catch (error) {
      appInstance.debug(`Failed to save default configuration: ${error}`);
    }

    return defaultConfig;
  }

  // If file exists but couldn't be parsed, return empty config to avoid data loss
  appInstance.debug(
    'Config file exists but could not be parsed, returning empty config to avoid data loss'
  );
  return {
    paths: [],
    commands: [],
  };
}

// Get default configuration for first-time installation
function getDefaultWebAppConfig(): WebAppPathConfig {
  const defaultCommands: CommandConfig[] = [
    {
      command: 'captureMoored',
      path: 'commands.captureMoored',
      registered: 'COMPLETED',
      description: 'Capture data when moored (position and wind)',
      active: false,
    },
  ];

  const defaultPaths: PathConfig[] = [
    {
      path: 'commands.captureMoored' as Path,
      name: 'Command: captureMoored',
      enabled: true,
      regimen: 'commands',
      source: undefined,
      context: 'vessels.self' as Context,
    },
    {
      path: 'navigation.position' as Path,
      name: 'Navigation Position',
      enabled: true,
      regimen: 'captureMoored',
      source: undefined,
      context: 'vessels.self' as Context,
    },
    {
      path: 'environment.wind.speedApparent' as Path,
      name: 'Apparent Wind Speed',
      enabled: true,
      regimen: 'captureMoored',
      source: undefined,
      context: 'vessels.self' as Context,
    },
  ];

  return {
    commands: defaultCommands,
    paths: defaultPaths,
  };
}

// Enhanced function to save webapp configuration
function saveWebAppConfig(
  paths: PathConfig[],
  commands: CommandConfig[]
): void {
  const webAppConfigPath = path.join(
    appInstance.getDataDirPath(),
    'signalk-parquet',
    'webapp-config.json'
  );
  try {
    const configDir = path.dirname(webAppConfigPath);
    if (!fs.existsSync(configDir)) {
      fs.mkdirSync(configDir, { recursive: true });
    }

    const webAppConfig: WebAppPathConfig = { paths, commands };
    fs.writeFileSync(webAppConfigPath, JSON.stringify(webAppConfig, null, 2));
    appInstance.debug(
      `Saved webapp configuration: ${paths.length} paths, ${commands.length} commands`
    );
  } catch (error) {
    appInstance.error(`Failed to save webapp configuration: ${error}`);
  }
}

// Legacy function for backward compatibility
function loadWebAppPaths(): PathConfig[] {
  return loadWebAppConfig().paths;
}

// Legacy function for backward compatibility
function saveWebAppPaths(paths: PathConfig[]): void {
  saveWebAppConfig(paths, currentCommands);
}

export default function (app: ServerAPI): SignalKPlugin {
  // Store app instance for global access
  appInstance = app;
  const plugin: SignalKPlugin = {
    id: 'signalk-parquet',
    name: 'SignalK to Parquet',
    description:
      'Save SignalK marine data directly to Parquet files with regimen-based control',
    schema: {},
    start: () => {},
    stop: () => {},
    registerWithRouter: undefined,
  };

  // Plugin state
  const state: PluginState = {
    unsubscribes: [],
    dataBuffers: new Map<string, DataRecord[]>(),
    activeRegimens: new Set<string>(),
    subscribedPaths: new Set<string>(),
    saveInterval: undefined,
    consolidationInterval: undefined,
    parquetWriter: undefined,
    s3Client: undefined,
    currentConfig: undefined,
    commandState: commandState,
  };

  plugin.start = async function (
    options: Partial<PluginConfig>
  ): Promise<void> {
    app.debug('Starting...');

    // Get vessel MMSI from SignalK
    const vesselMMSI =
      app.getSelfPath('mmsi') || app.getSelfPath('name') || 'unknown_vessel';

    // Use SignalK's application data directory
    const defaultOutputDir = path.join(app.getDataDirPath(), 'signalk-parquet');

    state.currentConfig = {
      bufferSize: options?.bufferSize || 1000,
      saveIntervalSeconds: options?.saveIntervalSeconds || 30,
      outputDirectory: options?.outputDirectory || defaultOutputDir,
      filenamePrefix: options?.filenamePrefix || 'signalk_data',
      retentionDays: options?.retentionDays || 7,
      fileFormat: options?.fileFormat || 'parquet',
      vesselMMSI: vesselMMSI,
      s3Upload: options?.s3Upload || { enabled: false },
    };

    // Load webapp configuration including commands
    const webAppConfig = loadWebAppConfig();
    currentPaths = webAppConfig.paths;
    currentCommands = webAppConfig.commands;

    // Initialize ParquetWriter
    state.parquetWriter = new ParquetWriter({
      format: state.currentConfig.fileFormat,
      app: app,
    });

    // Initialize S3 client if enabled
    app.debug(
      `DEBUG: S3 setup - enabled: ${state.currentConfig.s3Upload.enabled}, S3Client available: ${!!S3Client}`
    );
    if (state.currentConfig.s3Upload.enabled) {
      // Wait for AWS SDK import to complete
      try {
        if (!S3Client) {
          app.debug(`DEBUG: Waiting for AWS SDK import...`);
          const awsS3 = await import('@aws-sdk/client-s3');
          S3Client = awsS3.S3Client;
          PutObjectCommand = awsS3.PutObjectCommand;
          ListObjectsV2Command = awsS3.ListObjectsV2Command;
          HeadObjectCommand = awsS3.HeadObjectCommand;
          app.debug(`DEBUG: AWS SDK imported successfully`);
        }
      } catch (importError) {
        app.debug(`DEBUG: Failed to import AWS SDK: ${importError}`);
        S3Client = undefined;
      }
    }

    if (state.currentConfig.s3Upload.enabled && S3Client) {
      try {
        const s3Config: {
          region: string;
          credentials?: { accessKeyId: string; secretAccessKey: string };
        } = {
          region: state.currentConfig.s3Upload.region || 'us-east-1',
        };

        // Add credentials if provided
        if (
          state.currentConfig.s3Upload.accessKeyId &&
          state.currentConfig.s3Upload.secretAccessKey
        ) {
          s3Config.credentials = {
            accessKeyId: state.currentConfig.s3Upload.accessKeyId,
            secretAccessKey: state.currentConfig.s3Upload.secretAccessKey,
          };
          app.debug(`DEBUG: S3 credentials provided`);
        } else {
          app.debug(
            `DEBUG: No S3 credentials provided, using default AWS profile`
          );
        }

        app.debug(`DEBUG: Creating S3 client with region: ${s3Config.region}`);
        state.s3Client = new S3Client(s3Config);
        app.debug(
          `S3 client initialized for bucket: ${state.currentConfig.s3Upload.bucket}`
        );
      } catch (error) {
        app.debug(`Error initializing S3 client: ${error}`);
        state.s3Client = undefined;
      }
    } else {
      app.debug(
        `DEBUG: S3 client not initialized - enabled: ${state.currentConfig.s3Upload.enabled}, S3Client: ${!!S3Client}`
      );
    }

    // Ensure output directory exists
    fs.ensureDirSync(state.currentConfig.outputDirectory);

    // Subscribe to command paths first (these control regimens)
    subscribeToCommandPaths(state.currentConfig);

    // Check current command values at startup
    initializeRegimenStates(state.currentConfig);

    // Initialize command state
    initializeCommandState();

    // Subscribe to data paths based on initial regimen states
    updateDataSubscriptions(state.currentConfig);

    // Set up periodic save
    state.saveInterval = setInterval(() => {
      saveAllBuffers(state.currentConfig!);
    }, state.currentConfig.saveIntervalSeconds * 1000);

    // Set up daily consolidation (run at midnight UTC)
    const now = new Date();
    const nextMidnightUTC = new Date(
      Date.UTC(
        now.getUTCFullYear(),
        now.getUTCMonth(),
        now.getUTCDate() + 1,
        0,
        0,
        0,
        0
      )
    );
    const msUntilMidnightUTC = nextMidnightUTC.getTime() - now.getTime();

    app.debug(
      `Next consolidation at ${nextMidnightUTC.toISOString()} (in ${Math.round(msUntilMidnightUTC / 1000 / 60)} minutes)`
    );

    setTimeout(() => {
      consolidateYesterday(state.currentConfig!);

      // Then run daily consolidation every 24 hours
      state.consolidationInterval = setInterval(
        () => {
          consolidateYesterday(state.currentConfig!);
        },
        24 * 60 * 60 * 1000
      );
    }, msUntilMidnightUTC);

    // Run startup consolidation for missed previous days
    setTimeout(() => {
      consolidateMissedDays(state.currentConfig!);
    }, 5000); // Wait 5 seconds after startup to avoid conflicts

    // Upload all existing consolidated files to S3 (for catching up after BigInt fix)
    if (state.currentConfig.s3Upload.enabled) {
      setTimeout(() => {
        uploadAllConsolidatedFilesToS3(state.currentConfig!);
      }, 10000); // Wait 10 seconds after startup to avoid conflicts
    }

    registerHistoryApiRoute(
      app as unknown as Router,
      app.selfId,
      state.currentConfig?.outputDirectory || 'data',
      app.debug
    );
    app.debug('Started');
  };

  plugin.stop = function (): void {
    app.debug('Stopping...');

    // Clear intervals
    if (state.saveInterval) {
      clearInterval(state.saveInterval);
    }
    if (state.consolidationInterval) {
      clearInterval(state.consolidationInterval);
    }

    // Save any remaining buffered data
    saveAllBuffers();

    // Unsubscribe from all paths
    state.unsubscribes.forEach(unsubscribe => {
      if (typeof unsubscribe === 'function') {
        unsubscribe();
      }
    });
    state.unsubscribes = [];

    // Clean up stream subscriptions (new streambundle approach)
    if (state.streamSubscriptions) {
      state.streamSubscriptions.forEach(stream => {
        if (stream && typeof stream.end === 'function') {
          stream.end();
        }
      });
      state.streamSubscriptions = [];
    }

    // Clear data structures
    state.dataBuffers.clear();
    state.activeRegimens.clear();
    state.subscribedPaths.clear();
  };

  // Command Management Functions

  // Initialize command state on plugin start
  function initializeCommandState(): void {
    // Clear existing command state
    commandState.registeredCommands.clear();
    commandState.putHandlers.clear();

    // Re-register commands from configuration
    currentCommands.forEach((commandConfig: CommandConfig) => {
      const result = registerCommand(
        commandConfig.command,
        commandConfig.description
      );
      if (result.state === 'COMPLETED') {
        app.debug(`✅ Restored command: ${commandConfig.command}`);
      } else {
        app.error(
          `❌ Failed to restore command: ${commandConfig.command} - ${result.message}`
        );
      }
    });

    // Ensure all commands have path configurations (for backwards compatibility)
    let addedMissingPaths = false;
    currentCommands.forEach((commandConfig: CommandConfig) => {
      const commandPath = `commands.${commandConfig.command}`;
      const existingCommandPath = currentPaths.find(
        p => p.path === commandPath
      );
      if (!existingCommandPath) {
        const commandPathConfig: PathConfig = {
          path: commandPath as Path,
          name: `Command: ${commandConfig.command}`,
          enabled: true,
          regimen: undefined,
          source: undefined,
          context: 'vessels.self' as Context,
          excludeMMSI: undefined,
        };
        currentPaths.push(commandPathConfig);
        addedMissingPaths = true;
        app.debug(
          `✅ Added missing path configuration for existing command: ${commandConfig.command}`
        );
      }
    });

    // Save the updated configuration if we added missing paths
    if (addedMissingPaths) {
      saveWebAppConfig(currentPaths, currentCommands);
    }

    // Reset all commands to false on startup
    currentCommands.forEach((commandConfig: CommandConfig) => {
      initializeCommandValue(commandConfig.command, false);
    });

    app.debug(
      `🎮 Command state initialized with ${currentCommands.length} commands`
    );
  }

  // Command registration with full type safety
  function registerCommand(
    commandName: string,
    description?: string
  ): CommandExecutionResult {
    try {
      // Validate command name
      if (!isValidCommandName(commandName)) {
        return {
          state: 'FAILED',
          statusCode: 400,
          message:
            'Invalid command name. Use alphanumeric characters and underscores only.',
          timestamp: new Date().toISOString(),
        };
      }

      // Check if command already exists
      if (commandState.registeredCommands.has(commandName)) {
        return {
          state: 'FAILED',
          statusCode: 409,
          message: `Command '${commandName}' already registered`,
          timestamp: new Date().toISOString(),
        };
      }

      const commandPath = `commands.${commandName}`;
      const fullPath = `vessels.self.${commandPath}`;

      // Create command configuration
      const commandConfig: CommandConfig = {
        command: commandName,
        path: fullPath,
        registered: new Date().toISOString(),
        description: description || `Command: ${commandName}`,
        active: false,
        lastExecuted: undefined,
      };

      // Create PUT handler with proper typing
      const putHandler: CommandPutHandler = (
        context: string,
        path: string,
        // eslint-disable-next-line @typescript-eslint/no-explicit-any
        value: any,
        _callback?: (result: CommandExecutionResult) => void
      ): CommandExecutionResult => {
        app.debug(
          `Handling PUT for commands.${commandName} with value: ${JSON.stringify(value)}`
        );
        return executeCommand(commandName, Boolean(value));
      };

      // Register PUT handler with SignalK
      app.registerPutHandler(
        'vessels.self',
        commandPath,
        putHandler as unknown as () => void, //FIXME server api registerPutHandler is incorrectly typed https://github.com/SignalK/signalk-server/pull/2043
        'zennora-parquet-commands'
      );

      // Store command and handler
      commandState.registeredCommands.set(commandName, commandConfig);
      commandState.putHandlers.set(commandName, putHandler);

      // Initialize command value to false
      initializeCommandValue(commandName, false);

      // Update current commands and save
      currentCommands = Array.from(commandState.registeredCommands.values());

      // Automatically create a path configuration for this command
      const commandPathConfig: PathConfig = {
        path: commandPath as Path,
        name: `Command: ${commandName}`,
        enabled: true, // Always enabled so it gets subscribed to
        regimen: undefined, // Commands don't need regimen control
        source: undefined,
        context: 'vessels.self' as Context,
        excludeMMSI: undefined,
      };

      // Check if this command path already exists
      const existingCommandPath = currentPaths.find(
        p => p.path === commandPath
      );
      if (!existingCommandPath) {
        currentPaths.push(commandPathConfig);
        app.debug(
          `✅ Auto-created path configuration for command: ${commandName}`
        );
      }

      saveWebAppConfig(currentPaths, currentCommands);

      // Log command registration
      addCommandHistoryEntry(commandName, 'REGISTER', undefined, true);

      app.debug(`✅ Registered command: ${commandName} at ${fullPath}`);

      return {
        state: 'COMPLETED',
        statusCode: 200,
        message: `Command '${commandName}' registered successfully with automatic path configuration`,
        timestamp: new Date().toISOString(),
      };
    } catch (error) {
      const errorMessage = `Failed to register command '${commandName}': ${error}`;
      app.error(errorMessage);

      return {
        state: 'FAILED',
        statusCode: 500,
        message: errorMessage,
        timestamp: new Date().toISOString(),
      };
    }
  }

  // Command unregistration with type safety
  function unregisterCommand(commandName: string): CommandExecutionResult {
    try {
      const commandConfig = commandState.registeredCommands.get(commandName);
      if (!commandConfig) {
        return {
          state: 'FAILED',
          statusCode: 404,
          message: `Command '${commandName}' not found`,
          timestamp: new Date().toISOString(),
        };
      }

      // Remove PUT handler (SignalK API doesn't have unregister, but we can track it)
      commandState.putHandlers.delete(commandName);
      commandState.registeredCommands.delete(commandName);

      // Remove the auto-created path configuration for this command
      const commandPath = `commands.${commandName}`;
      const commandPathIndex = currentPaths.findIndex(
        p => p.path === commandPath
      );
      if (commandPathIndex !== -1) {
        currentPaths.splice(commandPathIndex, 1);
        app.debug(
          `🗑️ Removed auto-created path configuration for command: ${commandName}`
        );
      }

      // Update current commands and save
      currentCommands = Array.from(commandState.registeredCommands.values());
      saveWebAppConfig(currentPaths, currentCommands);

      // Log command unregistration
      addCommandHistoryEntry(commandName, 'UNREGISTER', undefined, true);

      app.debug(`🗑️ Unregistered command: ${commandName}`);

      return {
        state: 'COMPLETED',
        statusCode: 200,
        message: `Command '${commandName}' unregistered successfully with path cleanup`,
        timestamp: new Date().toISOString(),
      };
    } catch (error) {
      const errorMessage = `Failed to unregister command '${commandName}': ${error}`;
      app.error(errorMessage);

      return {
        state: 'FAILED',
        statusCode: 500,
        message: errorMessage,
        timestamp: new Date().toISOString(),
      };
    }
  }

  // Command execution with full type safety
  function executeCommand(
    commandName: string,
    value: boolean
  ): CommandExecutionResult {
    try {
      const commandConfig = commandState.registeredCommands.get(commandName);
      if (!commandConfig) {
        return {
          state: 'FAILED',
          statusCode: 404,
          message: `Command '${commandName}' not found`,
          timestamp: new Date().toISOString(),
        };
      }

      // Execute the command by sending delta
      const timestamp = new Date().toISOString();
      const delta: Delta = {
        context: 'vessels.self' as Context,
        updates: [
          {
            $source: 'signalk-parquet-commands' as SourceRef,
            timestamp: timestamp as Timestamp,
            values: [
              {
                path: `commands.${commandName}` as Path,
                value: value,
              },
            ],
          },
        ],
      };

      // Send delta message
      //FIXME see if delta can be Delta from the beginning
      app.handleMessage('signalk-parquet', delta as Delta);

      // Update command state
      commandConfig.active = value;
      commandConfig.lastExecuted = timestamp;
      commandState.registeredCommands.set(commandName, commandConfig);

      // Update current commands and save
      currentCommands = Array.from(commandState.registeredCommands.values());
      saveWebAppConfig(currentPaths, currentCommands);

      // Log command execution
      addCommandHistoryEntry(commandName, 'EXECUTE', value, true);

      app.debug(`🎮 Executed command: ${commandName} = ${value}`);

      return {
        state: 'COMPLETED',
        statusCode: 200,
        message: `Command '${commandName}' executed: ${value}`,
        timestamp: timestamp,
      };
    } catch (error) {
      const errorMessage = `Failed to execute command '${commandName}': ${error}`;
      app.error(errorMessage);

      addCommandHistoryEntry(
        commandName,
        'EXECUTE',
        value,
        false,
        errorMessage
      );

      return {
        state: 'FAILED',
        statusCode: 500,
        message: errorMessage,
        timestamp: new Date().toISOString(),
      };
    }
  }

  // Helper functions with type safety
  function isValidCommandName(commandName: string): boolean {
    const validPattern = /^[a-zA-Z0-9_]+$/;
    return (
      validPattern.test(commandName) &&
      commandName.length > 0 &&
      commandName.length <= 50
    );
  }

  function initializeCommandValue(commandName: string, value: boolean): void {
    const timestamp = new Date().toISOString() as Timestamp;
    const delta: Delta = {
      context: 'vessels.self' as Context,
      updates: [
        {
          $source: 'signalk-parquet-commands' as SourceRef,
          timestamp,
          values: [
            {
              path: `commands.${commandName}` as Path,
              value,
            },
          ],
        },
      ],
    };

    //FIXME
    app.handleMessage('signalk-parquet', delta as Delta);
  }

  function addCommandHistoryEntry(
    command: string,
    action: 'EXECUTE' | 'STOP' | 'REGISTER' | 'UNREGISTER',
    value?: boolean,
    success: boolean = true,
    error?: string
  ): void {
    const entry: CommandHistoryEntry = {
      command,
      action,
      value,
      timestamp: new Date().toISOString(),
      success,
      error,
    };

    commandHistory.push(entry);

    // Keep only last 100 entries
    if (commandHistory.length > 100) {
      commandHistory = commandHistory.slice(-100);
    }
  }

  // Subscribe to command paths that control regimens using proper subscription manager
  function subscribeToCommandPaths(config: PluginConfig): void {
    const commandPaths = currentPaths.filter(
      (pathConfig: PathConfig) =>
        pathConfig &&
        pathConfig.path &&
        pathConfig.path.startsWith('commands.') &&
        pathConfig.enabled
    );

    if (commandPaths.length === 0) return;

    const commandSubscription = {
      context: 'vessels.self' as Context,
      subscribe: commandPaths.map((pathConfig: PathConfig) => ({
        path: pathConfig.path,
        period: 1000, // Check commands every second
        policy: 'fixed' as const,
      })),
    };

    app.debug(
      `Subscribing to ${commandPaths.length} command paths via subscription manager`
    );

    app.subscriptionmanager.subscribe(
      commandSubscription,
      state.unsubscribes,
      (subscriptionError: unknown) => {
        app.debug(`Command subscription error: ${subscriptionError}`);
      },
      (delta: Delta) => {
        // Process each update in the delta
        if (delta.updates) {
          delta.updates.forEach((update: Update) => {
            if (hasValues(update)) {
              update.values.forEach((valueUpdate: PathValue) => {
                const pathConfig = commandPaths.find(
                  p => p.path === valueUpdate.path
                );
                if (pathConfig) {
                  handleCommandMessage(valueUpdate, pathConfig, config, update);
                }
              });
            }
          });
        }
      }
    );

    commandPaths.forEach(pathConfig => {
      state.subscribedPaths.add(pathConfig.path);
    });
  }

  // Handle command messages (regimen control) - now receives complete delta structure
  function handleCommandMessage(
    valueUpdate: PathValue,
    pathConfig: PathConfig,
    config: PluginConfig,
    update: Update
  ): void {
    try {
      app.debug(
        `📦 Received command update for ${pathConfig.path}: ${JSON.stringify(valueUpdate, null, 2)}`
      );

      // Check source filter if specified for commands too
      if (pathConfig.source && pathConfig.source.trim() !== '') {
        const messageSource =
          update.$source || (update.source ? update.source.label : null);
        if (messageSource !== pathConfig.source.trim()) {
          app.debug(
            `🚫 Command from source "${messageSource}" filtered out (expecting "${pathConfig.source.trim()}")`
          );
          return;
        }
      }

      if (valueUpdate.value !== undefined) {
        const commandName = extractCommandName(pathConfig.path);
        const isActive = Boolean(valueUpdate.value);

        app.debug(
          `Command ${commandName}: ${isActive ? 'ACTIVE' : 'INACTIVE'}`
        );

        if (isActive) {
          state.activeRegimens.add(commandName);
        } else {
          state.activeRegimens.delete(commandName);
        }

        // Debug active regimens state
        app.debug(
          `🎯 Active regimens: [${Array.from(state.activeRegimens).join(', ')}]`
        );

        // Update data subscriptions based on new regimen state
        updateDataSubscriptions(config);

        // Buffer this command change with complete metadata
        const bufferKey = `${pathConfig.context || 'vessels.self'}:${pathConfig.path}`;
        bufferData(
          bufferKey,
          {
            received_timestamp: new Date().toISOString(),
            signalk_timestamp: update.timestamp || new Date().toISOString(),
            context: 'vessels.self',
            path: valueUpdate.path,
            value: valueUpdate.value,
            source: update.source ? JSON.stringify(update.source) : undefined,
            source_label:
              update.$source ||
              (update.source ? update.source.label : undefined),
            source_type: update.source ? update.source.type : undefined,
            source_pgn: update.source ? update.source.pgn : undefined,
            source_src: update.source ? update.source.src : undefined,
            //FIXME
            // meta: valueUpdate.meta
            //   ? JSON.stringify(valueUpdate.meta)
            //   : undefined,
          },
          config
        );
      }
    } catch (error) {
      app.debug(`Error handling command message: ${error}`);
    }
  }

  // Helper function to handle wildcard contexts
  function handleWildcardContext(pathConfig: PathConfig): PathConfig {
    const context = pathConfig.context || 'vessels.self';

    if (context === 'vessels.*') {
      // For vessels.*, we create a subscription that will receive deltas from any vessel
      // The actual filtering by MMSI will happen in the delta handler
      return {
        ...pathConfig,
        context: 'vessels.*' as Context, // Keep the wildcard for the subscription
      };
    }

    // Not a wildcard, return as-is
    return pathConfig;
  }

  // Helper function to check if a vessel should be excluded based on MMSI
  function shouldExcludeVessel(
    vesselContext: string,
    pathConfig: PathConfig
  ): boolean {
    if (!pathConfig.excludeMMSI || pathConfig.excludeMMSI.length === 0) {
      return false; // No exclusions specified
    }

    try {
      // For vessels.self, use getSelfPath
      if (vesselContext === 'vessels.self') {
        const mmsiData = app.getSelfPath('mmsi');
        if (mmsiData && mmsiData.value) {
          const mmsi = String(mmsiData.value);
          return pathConfig.excludeMMSI.includes(mmsi);
        }
      } else {
        // For other vessels, we would need to get their MMSI from the delta or other means
        // For now, we'll skip MMSI filtering for other vessels
        app.debug(
          `MMSI filtering not implemented for vessel context: ${vesselContext}`
        );
      }
    } catch (error) {
      app.debug(`Error checking MMSI for vessel ${vesselContext}: ${error}`);
    }

    return false; // Don't exclude if we can't determine MMSI
  }

  // Update data path subscriptions based on active regimens
  function updateDataSubscriptions(config: PluginConfig): void {
    // First, unsubscribe from all existing subscriptions
    state.unsubscribes.forEach(unsubscribe => {
      if (typeof unsubscribe === 'function') {
        unsubscribe();
      }
    });
    state.unsubscribes = [];
    state.subscribedPaths.clear();

    app.debug('Cleared all existing subscriptions');

    // Re-subscribe to command paths
    subscribeToCommandPaths(config);

    // Now subscribe to data paths using currentPaths
    const dataPaths = currentPaths.filter(
      (pathConfig: PathConfig) =>
        pathConfig &&
        pathConfig.path &&
        !pathConfig.path.startsWith('commands.')
    );

    const shouldSubscribePaths = dataPaths.filter((pathConfig: PathConfig) =>
      shouldSubscribeToPath(pathConfig)
    );

    // Handle wildcard contexts (like vessels.*)
    const processedPaths: PathConfig[] = shouldSubscribePaths.map(pathConfig =>
      handleWildcardContext(pathConfig)
    );

    if (processedPaths.length === 0) {
      app.debug('No data paths need subscription currently');
      return;
    }

    // Group paths by context for separate subscriptions
    const contextGroups = new Map<Context, PathConfig[]>();
    processedPaths.forEach((pathConfig: PathConfig) => {
      const context = (pathConfig.context || 'vessels.self') as Context;
      if (!contextGroups.has(context)) {
        contextGroups.set(context, []);
      }
      contextGroups.get(context)!.push(pathConfig);
    });

    // Use app.streambundle approach as recommended by SignalK developer
    // This avoids server arbitration and provides true source filtering
    contextGroups.forEach((pathConfigs, context) => {
      app.debug(
        `Creating ${pathConfigs.length} streambundle subscriptions for context ${context}`
      );

      pathConfigs.forEach((pathConfig: PathConfig) => {
        // Show MMSI exclusion config for troubleshooting
        if (pathConfig.excludeMMSI && pathConfig.excludeMMSI.length > 0) {
          app.debug(
            `🔧 Path ${pathConfig.path} has MMSI exclusions: [${pathConfig.excludeMMSI.join(', ')}]`
          );
        }

        // Create individual stream for each path (developer's recommended approach)
        const stream = app.streambundle
          .getBus(pathConfig.path as Path)
          .filter((normalizedDelta: NormalizedDelta) => {
            // Filter by source if specified
            if (pathConfig.source && pathConfig.source.trim() !== '') {
              const expectedSource = pathConfig.source.trim();
              const actualSource = normalizedDelta.$source;

              if (actualSource !== expectedSource) {
                return false;
              }
            }

            // Filter by context
            const targetContext = pathConfig.context || 'vessels.self';
            if (targetContext === 'vessels.*') {
              // For wildcard, accept any vessel context
              if (!normalizedDelta.context.startsWith('vessels.')) {
                return false;
              }
            } else if (targetContext === 'vessels.self') {
              // For vessels.self, check if this is the server's own vessel
              const selfContext = app.selfContext;
              const selfVessel = app.getSelfPath('') || {};
              const selfMMSI = selfVessel.mmsi;
              const selfUuid = app.getSelfPath('uuid');

              // Check if the context matches the server's self vessel
              let isSelfVessel = false;

              if (normalizedDelta.context === 'vessels.self') {
                isSelfVessel = true;
              } else if (normalizedDelta.context === selfContext) {
                isSelfVessel = true;
              } else if (
                selfMMSI &&
                normalizedDelta.context.includes(selfMMSI)
              ) {
                isSelfVessel = true;
              } else if (
                selfUuid &&
                normalizedDelta.context.includes(selfUuid)
              ) {
                isSelfVessel = true;
              }

              if (!isSelfVessel) {
                return false;
              }
            } else {
              // For specific context, match exactly
              if (normalizedDelta.context !== targetContext) {
                return false;
              }
            }

            // MMSI exclusion filtering
            if (pathConfig.excludeMMSI && pathConfig.excludeMMSI.length > 0) {
              const contextHasExcludedMMSI = pathConfig.excludeMMSI.some(mmsi =>
                normalizedDelta.context.includes(mmsi)
              );
              if (contextHasExcludedMMSI) {
                return false;
              }
            }

            return true;
          })
          .debounceImmediate(1000) // Built-in debouncing as recommended
          .onValue((normalizedDelta: NormalizedDelta) => {
            handleStreamData(normalizedDelta, pathConfig, config);
          });

        // Store stream reference for cleanup (instead of unsubscribe functions)
        state.streamSubscriptions = state.streamSubscriptions || [];
        state.streamSubscriptions.push(stream);
        state.subscribedPaths.add(pathConfig.path);
      });
    });
  }

  // Determine if we should subscribe to a path based on regimens
  function shouldSubscribeToPath(pathConfig: PathConfig): boolean {
    // Always subscribe if explicitly enabled
    if (pathConfig.enabled) {
      app.debug(`✅ Path ${pathConfig.path} enabled (always on)`);
      return true;
    }

    // Check if any required regimens are active
    if (pathConfig.regimen) {
      const requiredRegimens = pathConfig.regimen.split(',').map(r => r.trim());
      const hasActiveRegimen = requiredRegimens.some(regimen =>
        state.activeRegimens.has(regimen)
      );
      app.debug(
        `🔍 Path ${pathConfig.path} requires regimens [${requiredRegimens.join(', ')}], active: [${Array.from(state.activeRegimens).join(', ')}] → ${hasActiveRegimen ? 'SUBSCRIBE' : 'SKIP'}`
      );
      return hasActiveRegimen;
    }

    app.debug(
      `❌ Path ${pathConfig.path} has no regimen control and not enabled`
    );
    return false;
  }

  // Handle data messages from SignalK - now receives complete delta structure
  // Legacy handler - kept for rollback capability
  // eslint-disable-next-line @typescript-eslint/no-unused-vars
  function handleDataMessage(
    valueUpdate: PathValue,
    pathConfig: PathConfig,
    config: PluginConfig,
    update: Update,
    delta: Delta
  ): void {
    try {
      // Check if this vessel should be excluded based on MMSI
      const vesselContext = delta.context || 'vessels.self';
      if (shouldExcludeVessel(vesselContext, pathConfig)) {
        app.debug(
          `Excluding data from vessel ${vesselContext} due to MMSI filter`
        );
        return;
      }

      // Check source filter if specified
      if (pathConfig.source && pathConfig.source.trim() !== '') {
        const messageSource =
          update.$source || (update.source ? update.source.label : null);
        if (messageSource !== pathConfig.source.trim()) {
          // Source doesn't match filter, skip this message
          return;
        }
      }

      // Retrieve metadata for this path
      let metadata: string | undefined;
      try {
        // eslint-disable-next-line @typescript-eslint/no-explicit-any
        const pathMetadata = (app as any).getMetadata?.(valueUpdate.path);
        if (pathMetadata) {
          metadata = JSON.stringify(pathMetadata);
        }
      } catch (error) {
        // Metadata retrieval failed, continue without it
        app.debug(
          `Failed to retrieve metadata for ${valueUpdate.path}: ${error}`
        );
      }

      const record: DataRecord = {
        received_timestamp: new Date().toISOString(),
        signalk_timestamp: update.timestamp || new Date().toISOString(),
        context: delta.context || pathConfig.context || 'vessels.self', // Use actual context from delta message
        path: valueUpdate.path,
        value: null,
        value_json: undefined,
        source: update.source ? JSON.stringify(update.source) : undefined,
        source_label:
          update.$source || (update.source ? update.source.label : undefined),
        source_type: update.source ? update.source.type : undefined,
        source_pgn: update.source ? update.source.pgn : undefined,
        source_src: update.source ? update.source.src : undefined,
        meta:
          metadata ||
          // eslint-disable-next-line @typescript-eslint/no-explicit-any
          ((valueUpdate as any).meta
            ? // eslint-disable-next-line @typescript-eslint/no-explicit-any
              JSON.stringify((valueUpdate as any).meta)
            : undefined),
      };

      // Handle different value types (matching Python logic)
      if (typeof valueUpdate.value === 'object' && valueUpdate.value !== null) {
        record.value_json = JSON.stringify(valueUpdate.value);

        // Flatten object properties for easier querying
        for (const [key, val] of Object.entries(valueUpdate.value)) {
          if (
            typeof val === 'string' ||
            typeof val === 'number' ||
            typeof val === 'boolean'
          ) {
            // eslint-disable-next-line @typescript-eslint/no-explicit-any
            (record as any)[`value_${key}`] = val;
          }
        }
      } else {
        record.value = valueUpdate.value;
      }

      // Use actual context + path as buffer key to separate data from different vessels
      const actualContext =
        delta.context || pathConfig.context || 'vessels.self';

      const bufferKey = `${actualContext}:${pathConfig.path}`;
      bufferData(bufferKey, record, config);
    } catch (error) {
      app.debug(`Error handling data message: ${error}`);
    }
  }

  // New handler for streambundle data (developer's recommended approach)
  function handleStreamData(
    normalizedDelta: NormalizedDelta,
    pathConfig: PathConfig,
    config: PluginConfig
  ): void {
    try {
      // Retrieve metadata for this path
      let metadata: string | undefined;
      try {
        // eslint-disable-next-line @typescript-eslint/no-explicit-any
        const pathMetadata = (app as any).getMetadata?.(normalizedDelta.path);
        if (pathMetadata) {
          metadata = JSON.stringify(pathMetadata);
        }
      } catch (error) {
        // Metadata retrieval failed, continue without it
        app.debug(
          `Failed to retrieve metadata for ${normalizedDelta.path}: ${error}`
        );
      }

      const record: DataRecord = {
        received_timestamp: new Date().toISOString(),
        signalk_timestamp:
          normalizedDelta.timestamp || new Date().toISOString(),
        context:
          normalizedDelta.context || pathConfig.context || 'vessels.self',
        path: normalizedDelta.path,
        value: null,
        value_json: undefined,
        source: normalizedDelta.source
          ? JSON.stringify(normalizedDelta.source)
          : undefined,
        source_label: normalizedDelta.$source || undefined,
        source_type: normalizedDelta.source
          ? normalizedDelta.source.type
          : undefined,
        source_pgn: normalizedDelta.source
          ? normalizedDelta.source.pgn
          : undefined,
        source_src: normalizedDelta.source
          ? normalizedDelta.source.src
          : undefined,
        meta: metadata,
      };

      // Handle complex values
      if (
        typeof normalizedDelta.value === 'object' &&
        normalizedDelta.value !== null
      ) {
        record.value_json = JSON.stringify(normalizedDelta.value);
        // Extract key properties as columns for easier querying
        Object.entries(normalizedDelta.value).forEach(([key, val]) => {
          if (
            typeof val === 'string' ||
            typeof val === 'number' ||
            typeof val === 'boolean'
          ) {
            // eslint-disable-next-line @typescript-eslint/no-explicit-any
            (record as any)[`value_${key}`] = val;
          }
        });
      } else {
        record.value = normalizedDelta.value;
      }

      // Use actual context + path as buffer key to separate data from different vessels
      const bufferKey = `${normalizedDelta.context}:${pathConfig.path}`;
      bufferData(bufferKey, record, config);
    } catch (error) {
      app.debug(`Error handling stream data: ${error}`);
    }
  }

  // Buffer data and trigger save if buffer is full
  function bufferData(
    signalkPath: string,
    record: DataRecord,
    config: PluginConfig
  ): void {
    if (!state.dataBuffers.has(signalkPath)) {
      state.dataBuffers.set(signalkPath, []);
    }

    const buffer = state.dataBuffers.get(signalkPath)!;
    buffer.push(record);

    if (buffer.length >= config.bufferSize) {
      // Extract the actual SignalK path from the buffer key (context:path format)
      // Find the separator between context and path - look for the last colon followed by a valid SignalK path
      const pathMatch = signalkPath.match(/^.*:([a-zA-Z][a-zA-Z0-9._]*)$/);
      const actualPath = pathMatch ? pathMatch[1] : signalkPath;
      const urnMatch = signalkPath.match(/^([^:]+):/);
      const urn = urnMatch ? urnMatch[1] : 'vessels.self';
      app.debug(
        `💾 Buffer full: ${buffer.length} rows | ${urn} | ${actualPath} | trigger=buffer_full`
      );
      saveBufferToParquet(actualPath, buffer, config);
      state.dataBuffers.set(signalkPath, []); // Clear buffer
    }
  }

  // Save all buffers (called periodically and on shutdown)
  function saveAllBuffers(config?: PluginConfig): void {
    state.dataBuffers.forEach((buffer, signalkPath) => {
      if (buffer.length > 0) {
        // Extract the actual SignalK path from the buffer key (context:path format)
        // Find the separator between context and path - look for the last colon followed by a valid SignalK path
        const pathMatch = signalkPath.match(/^.*:([a-zA-Z][a-zA-Z0-9._]*)$/);
        const actualPath = pathMatch ? pathMatch[1] : signalkPath;
        const urnMatch = signalkPath.match(/^([^:]+):/);
        const urn = urnMatch ? urnMatch[1] : 'vessels.self';
        app.debug(
          `💾 Timer flush: ${buffer.length} rows | ${urn} | ${actualPath} | trigger=timer`
        );
        saveBufferToParquet(actualPath, buffer, config || state.currentConfig!);
        state.dataBuffers.set(signalkPath, []); // Clear buffer
      }
    });
  }

  // Save buffer to Parquet file
  async function saveBufferToParquet(
    signalkPath: string,
    buffer: DataRecord[],
    config: PluginConfig
  ): Promise<void> {
    try {
      // Get context from first record in buffer (all records in buffer have same path/context)
      const context = buffer.length > 0 ? buffer[0].context : 'vessels.self';

      // Create proper directory structure
      let contextPath: string;
      if (context === 'vessels.self') {
        // Clean the self context for filesystem usage (replace dots with slashes, colons with underscores)
        contextPath = app.selfContext.replace(/\./g, '/').replace(/:/g, '_');
      } else if (context.startsWith('vessels.')) {
        // Extract vessel identifier and clean it for filesystem
        const vesselId = context.replace('vessels.', '').replace(/:/g, '_');
        contextPath = `vessels/${vesselId}`;
      } else if (context.startsWith('meteo.')) {
        // Extract meteo station identifier and clean it for filesystem
        const meteoId = context.replace('meteo.', '').replace(/:/g, '_');
        contextPath = `meteo/${meteoId}`;
      } else {
        // Fallback: clean the entire context
        contextPath = context.replace(/:/g, '_').replace(/\./g, '/');
      }

      const dirPath = path.join(
        config.outputDirectory,
        contextPath,
        signalkPath.replace(/\./g, '/')
      );
      await fs.ensureDir(dirPath);

      // Generate filename with timestamp
      const timestamp = new Date()
        .toISOString()
        .replace(/[:.]/g, '')
        .slice(0, 15);
      const fileExt =
        config.fileFormat === 'csv'
          ? 'csv'
          : config.fileFormat === 'parquet'
            ? 'parquet'
            : 'json';
      const filename = `${config.filenamePrefix}_${timestamp}.${fileExt}`;
      const filepath = path.join(dirPath, filename);

      // Use ParquetWriter to save in the configured format
      const savedPath = await state.parquetWriter!.writeRecords(
        filepath,
        buffer
      );

      // Upload to S3 if enabled and timing is real-time
      if (config.s3Upload.enabled && config.s3Upload.timing === 'realtime') {
        await uploadToS3(savedPath, config);
      }
    } catch (error) {
      app.debug(`❌ Error saving buffer for ${signalkPath}: ${error}`);
    }
  }

  // Startup consolidation for missed previous days (excludes current day)
  async function consolidateMissedDays(config: PluginConfig): Promise<void> {
    try {
      app.debug('Checking for missed consolidations at startup...');

      // Get list of all date directories that exist
      const outputDir = config.outputDirectory;
      if (!(await fs.pathExists(outputDir))) {
        return;
      }

      // Find all non-consolidated files older than today
      const today = new Date();
      today.setUTCHours(0, 0, 0, 0);

      const pattern = path.join(outputDir, '**/*.parquet');
      const files = await glob(pattern);

      // Extract dates from files and find days that need consolidation
      const datesNeedingConsolidation = new Set<string>();

      for (const file of files) {
        // Skip already consolidated files
        if (file.includes('_consolidated.parquet')) {
          continue;
        }

        // Extract date from filename (format: signalk_data_2025-07-14T1847.parquet)
        const filename = path.basename(file);
        const dateMatch = filename.match(
          /(\d{4})-(\d{2})-(\d{2})T\d{4}\.parquet$/
        );

        if (dateMatch) {
          const year = parseInt(dateMatch[1]);
          const month = parseInt(dateMatch[2]) - 1; // Month is 0-based
          const day = parseInt(dateMatch[3]);
          const fileDate = new Date(year, month, day);
          fileDate.setUTCHours(0, 0, 0, 0);

          // Only consolidate if file is from before today
          if (fileDate < today) {
            const dateStr = `${year}-${month + 1 < 10 ? '0' : ''}${month + 1}-${day < 10 ? '0' : ''}${day}`;
            datesNeedingConsolidation.add(dateStr);
          }
        }
      }

      // Consolidate each missed day
      for (const dateStr of datesNeedingConsolidation) {
        // Parse date string format: 2025-07-14
        const [yearStr, monthStr, dayStr] = dateStr.split('-');
        const date = new Date(
          parseInt(yearStr),
          parseInt(monthStr) - 1,
          parseInt(dayStr)
        );

        app.debug(`Consolidating missed day: ${dateStr}`);

        const consolidatedCount = await state.parquetWriter!.consolidateDaily(
          config.outputDirectory,
          date,
          config.filenamePrefix
        );

        app.debug(
          `DEBUG: consolidatedCount = ${consolidatedCount} for date ${dateStr}`
        );

        if (consolidatedCount > 0) {
          app.debug(
            `Consolidated ${consolidatedCount} topic directories for ${dateStr}`
          );

          // Upload consolidated files to S3 if enabled and timing is consolidation
          app.debug(
            `DEBUG: S3 check - enabled: ${config.s3Upload.enabled}, timing: ${config.s3Upload.timing}`
          );
          if (
            config.s3Upload.enabled &&
            config.s3Upload.timing === 'consolidation'
          ) {
            app.debug(`DEBUG: Starting S3 upload for ${dateStr}`);
            await uploadConsolidatedFilesToS3(config, date);
          } else {
            app.debug(`DEBUG: S3 upload skipped for ${dateStr}`);
          }
        }
      }

      if (datesNeedingConsolidation.size > 0) {
        app.debug(
          `Startup consolidation completed for ${datesNeedingConsolidation.size} missed days`
        );
      } else {
        app.debug('No missed consolidations found at startup');
      }
    } catch (error) {
      app.debug(`Error during startup consolidation: ${error}`);
    }
  }

  // Daily consolidation function
  async function consolidateYesterday(config: PluginConfig): Promise<void> {
    try {
      const yesterday = new Date();
      yesterday.setUTCDate(yesterday.getUTCDate() - 1);

      const consolidatedCount = await state.parquetWriter!.consolidateDaily(
        config.outputDirectory,
        yesterday,
        config.filenamePrefix
      );

      app.debug(
        `DEBUG: Daily consolidatedCount = ${consolidatedCount} for yesterday`
      );

      if (consolidatedCount > 0) {
        app.debug(
          `Consolidated ${consolidatedCount} topic directories for ${yesterday.toISOString().split('T')[0]}`
        );

        // Upload consolidated files to S3 if enabled and timing is consolidation
        if (
          config.s3Upload.enabled &&
          config.s3Upload.timing === 'consolidation'
        ) {
          await uploadConsolidatedFilesToS3(config, yesterday);
        }
      }
    } catch (error) {
      app.debug(`Error during daily consolidation: ${error}`);
    }
  }

  // Upload all existing consolidated files to S3 (for catching up after BigInt fix)
  async function uploadAllConsolidatedFilesToS3(
    config: PluginConfig
  ): Promise<void> {
    try {
      app.debug(`DEBUG: Uploading all existing consolidated files to S3`);

      // Find all consolidated parquet files
      const consolidatedPattern = `**/*_consolidated.parquet`;
      const consolidatedFiles = await glob(consolidatedPattern, {
        cwd: config.outputDirectory,
        absolute: true,
        nodir: true,
      });

      app.debug(
        `DEBUG: Found ${consolidatedFiles.length} consolidated files to upload`
      );

      let uploadedCount = 0;
      for (const filePath of consolidatedFiles) {
        const success = await uploadToS3(filePath, config);
        if (success) uploadedCount++;
      }

      app.debug(
        `DEBUG: Successfully uploaded ${uploadedCount}/${consolidatedFiles.length} consolidated files`
      );
    } catch (error) {
      app.debug(`Error uploading all consolidated files to S3: ${error}`);
    }
  }

  // Upload consolidated files to S3
  async function uploadConsolidatedFilesToS3(
    config: PluginConfig,
    date: Date
  ): Promise<void> {
    try {
      const dateStr = date.toISOString().split('T')[0];
      const consolidatedPattern = `**/*_${dateStr}_consolidated.parquet`;

      app.debug(`DEBUG: S3 upload function called for ${dateStr}`);
      app.debug(
        `DEBUG: Looking for pattern: ${consolidatedPattern} in ${config.outputDirectory}`
      );

      // Find all consolidated files for the date
      const consolidatedFiles = await glob(consolidatedPattern, {
        cwd: config.outputDirectory,
        absolute: true,
        nodir: true,
      });

      app.debug(
        `Found ${consolidatedFiles.length} consolidated files to upload for ${dateStr}`
      );
      app.debug(`DEBUG: Files found: ${JSON.stringify(consolidatedFiles)}`);

      // Upload each consolidated file
      for (const filePath of consolidatedFiles) {
        await uploadToS3(filePath, config);
      }
    } catch (error) {
      app.debug(`Error uploading consolidated files to S3: ${error}`);
    }
  }

  // Initialize regimen states from current API values at startup
  function initializeRegimenStates(_config: PluginConfig): void {
    const commandPaths = currentPaths.filter(
      (pathConfig: PathConfig) =>
        pathConfig &&
        pathConfig.path &&
        pathConfig.path.startsWith('commands.') &&
        pathConfig.enabled
    );

    app.debug(
      `🔍 Checking current command values for ${commandPaths.length} command paths at startup`
    );

    commandPaths.forEach((pathConfig: PathConfig) => {
      try {
        // Get current value from SignalK API
        const currentData = app.getSelfPath(pathConfig.path);

        if (currentData !== undefined && currentData !== null) {
          app.debug(
            `📋 Found current value for ${pathConfig.path}: ${JSON.stringify(currentData)}`
          );

          // Check if there's source information
          const shouldProcess = true;

          // If source filter is specified, check it
          if (pathConfig.source && pathConfig.source.trim() !== '') {
            // For startup, we need to check the API source info
            // This is a simplified check - in real deltas we get more source info
            app.debug(
              `🔍 Source filter specified for ${pathConfig.path}: "${pathConfig.source.trim()}"`
            );

            // For now, we'll process the value if it exists and log a warning
            // In practice, you might want to check the source here too
            app.debug(
              `⚠️  Startup value processed without source verification for ${pathConfig.path}`
            );
          }

          if (shouldProcess && currentData.value !== undefined) {
            const commandName = extractCommandName(pathConfig.path);
            const isActive = Boolean(currentData.value);

            app.debug(
              `🚀 Startup: Command ${commandName}: ${isActive ? 'ACTIVE' : 'INACTIVE'}`
            );

            if (isActive) {
              state.activeRegimens.add(commandName);
            } else {
              state.activeRegimens.delete(commandName);
            }
          }
        } else {
          app.debug(`📭 No current value found for ${pathConfig.path}`);
        }
      } catch (error) {
        app.debug(
          `❌ Error checking startup value for ${pathConfig.path}: ${error}`
        );
      }
    });

    app.debug(
      `🎯 Startup regimens initialized: [${Array.from(state.activeRegimens).join(', ')}]`
    );
  }

  // Helper functions
  function extractCommandName(signalkPath: string): string {
    // Extract command name from "commands.captureWeather"
    const parts = signalkPath.split('.');
    return parts[parts.length - 1];
  }

  // Convert BigInt values to regular numbers for JSON serialization
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  function mapForJSON(rawData: any[]): any[] {
    return rawData.map(row => {
      const convertedRow: typeof row = {};
      for (const [key, value] of Object.entries(row)) {
        convertedRow[key] = typeof value === 'bigint' ? Number(value) : value;
      }
      return convertedRow;
    });
  }

  // S3 upload function
  async function uploadToS3(
    filePath: string,
    config: PluginConfig
  ): Promise<boolean> {
    if (!config.s3Upload.enabled || !state.s3Client || !PutObjectCommand) {
      return false;
    }

    try {
      // Generate S3 key first
      const relativePath = path.relative(config.outputDirectory, filePath);
      let s3Key = relativePath;
      if (config.s3Upload.keyPrefix) {
        const prefix = config.s3Upload.keyPrefix.endsWith('/')
          ? config.s3Upload.keyPrefix
          : `${config.s3Upload.keyPrefix}/`;
        s3Key = `${prefix}${relativePath}`;
      }

      // Check if file exists in S3 and compare timestamps
      const localStats = await fs.stat(filePath);
      let shouldUpload = true;

      try {
        if (HeadObjectCommand) {
          const headCommand = new HeadObjectCommand({
            Bucket: config.s3Upload.bucket,
            Key: s3Key,
          });
          const s3Object = await state.s3Client.send(headCommand);

          if (s3Object.LastModified) {
            const s3LastModified = new Date(s3Object.LastModified);
            const localLastModified = new Date(localStats.mtime);

            if (localLastModified <= s3LastModified) {
              shouldUpload = false;
              app.debug(`⏭️ Skipping upload, S3 file is newer: ${s3Key}`);
            } else {
              app.debug(`📤 Local file is newer, uploading: ${s3Key}`);
            }
          }
        }
        // eslint-disable-next-line @typescript-eslint/no-explicit-any
      } catch (headError: any) {
        if (
          headError.name === 'NotFound' ||
          headError.$metadata?.httpStatusCode === 404
        ) {
          app.debug(`📤 File not found in S3, uploading: ${s3Key}`);
          shouldUpload = true;
        } else {
          app.debug(
            `⚠️ Error checking S3 file, uploading anyway: ${headError.message}`
          );
          shouldUpload = true;
        }
      }

      if (!shouldUpload) {
        return true; // Consider it successful since file is already up to date
      }

      // Read the file
      const fileContent = await fs.readFile(filePath);

      // Upload to S3
      const command = new PutObjectCommand({
        Bucket: config.s3Upload.bucket,
        Key: s3Key,
        Body: fileContent,
        ContentType: filePath.endsWith('.parquet')
          ? 'application/octet-stream'
          : 'application/json',
      });

      await state.s3Client.send(command);
      app.debug(`✅ Uploaded to S3: s3://${config.s3Upload.bucket}/${s3Key}`);

      // Delete local file if configured
      if (config.s3Upload.deleteAfterUpload) {
        await fs.unlink(filePath);
        app.debug(`🗑️ Deleted local file: ${filePath}`);
      }

      return true;
    } catch (error) {
      app.debug(`❌ Error uploading ${filePath} to S3: ${error}`);
      return false;
    }
  }

  plugin.schema = {
    type: 'object',
    title: 'SignalK to Parquet Data Store',
    description:
      "The archiving commands, paths and other processes are managed in the companion 'SignalK to Parquet Data Store' in the Webapp section.\n\nThese settings here underpin the system.",
    properties: {
      bufferSize: {
        type: 'number',
        title: 'Buffer Size',
        description: 'Number of records to buffer before writing to file',
        default: 1000,
        minimum: 10,
        maximum: 10000,
      },
      saveIntervalSeconds: {
        type: 'number',
        title: 'Save Interval (seconds)',
        description: 'How often to save buffered data to files',
        default: 30,
        minimum: 5,
        maximum: 300,
      },
      outputDirectory: {
        type: 'string',
        title: 'Output Directory',
        description:
          'Directory to save data files (defaults to application_data/{vessel}/signalk-parquet)',
        default: '',
      },
      filenamePrefix: {
        type: 'string',
        title: 'Filename Prefix',
        description: 'Prefix for generated filenames',
        default: 'signalk_data',
      },
      fileFormat: {
        type: 'string',
        title: 'File Format',
        description: 'Format for saved data files',
        enum: ['json', 'csv', 'parquet'],
        default: 'parquet',
      },
      retentionDays: {
        type: 'number',
        title: 'Retention Days',
        description: 'Days to keep processed files',
        default: 7,
        minimum: 1,
        maximum: 365,
      },
      s3Upload: {
        type: 'object',
        title: 'S3 Upload Configuration',
        description: 'Optional S3 backup/archive functionality',
        properties: {
          enabled: {
            type: 'boolean',
            title: 'Enable S3 Upload',
            description: 'Enable uploading files to Amazon S3',
            default: false,
          },
          timing: {
            type: 'string',
            title: 'Upload Timing',
            description: 'When to upload files to S3',
            enum: ['realtime', 'consolidation'],
            enumNames: [
              'Real-time (after each file save)',
              'At consolidation (daily)',
            ],
            default: 'consolidation',
          },
          bucket: {
            type: 'string',
            title: 'S3 Bucket Name',
            description: 'Name of the S3 bucket to upload to',
            default: '',
          },
          region: {
            type: 'string',
            title: 'AWS Region',
            description: 'AWS region where the S3 bucket is located',
            default: 'us-east-1',
          },
          keyPrefix: {
            type: 'string',
            title: 'S3 Key Prefix',
            description:
              'Optional prefix for S3 object keys (e.g., "marine-data/")',
            default: '',
          },
          accessKeyId: {
            type: 'string',
            title: 'AWS Access Key ID',
            description:
              'AWS Access Key ID (leave empty to use IAM role or environment variables)',
            default: '',
          },
          secretAccessKey: {
            type: 'string',
            title: 'AWS Secret Access Key',
            description:
              'AWS Secret Access Key (leave empty to use IAM role or environment variables)',
            default: '',
          },
          deleteAfterUpload: {
            type: 'boolean',
            title: 'Delete Local Files After Upload',
            description: 'Delete local files after successful upload to S3',
            default: false,
          },
        },
      },
    },
  };

  // Webapp static files and API routes
  plugin.registerWithRouter = function (router: Router): void {
    // Serve static files from public directory
    const publicPath = path.join(__dirname, '../public');
    if (fs.existsSync(publicPath)) {
      router.use(express.static(publicPath));
      app.debug(`Static files served from: ${publicPath}`);
    }

    // Get the current configuration for data directory
    const getDataDir = (): string => {
      // Use the user-configured output directory, fallback to SignalK default
      return state.currentConfig?.outputDirectory || app.getDataDirPath();
    };

    // Helper function to get available paths from directory structure
    function getAvailablePaths(dataDir: string): PathInfo[] {
      const paths: PathInfo[] = [];
      // Clean the self context for filesystem usage (replace dots with slashes, colons with underscores)
      const selfContextPath = app.selfContext
        .replace(/\./g, '/')
        .replace(/:/g, '_');
      const vesselsDir = path.join(dataDir, selfContextPath);

      app.debug(`🔍 Looking for paths in vessel directory: ${vesselsDir}`);
      app.debug(
        `📡 Using vessel context: ${app.selfContext} → ${selfContextPath}`
      );

      if (!fs.existsSync(vesselsDir)) {
        app.debug(`❌ Vessel directory does not exist: ${vesselsDir}`);
        return paths;
      }

      function walkPaths(currentPath: string, relativePath: string = ''): void {
        try {
          app.debug(
            `🚶 Walking path: ${currentPath} (relative: ${relativePath})`
          );
          const items = fs.readdirSync(currentPath);
          items.forEach((item: string) => {
            const fullPath = path.join(currentPath, item);
            const stat = fs.statSync(fullPath);

            if (
              stat.isDirectory() &&
              item !== 'processed' &&
              item !== 'failed'
            ) {
              const newRelativePath = relativePath
                ? `${relativePath}.${item}`
                : item;

              // Check if this directory has parquet files
              const hasParquetFiles = fs
                .readdirSync(fullPath)
                .some((file: string) => file.endsWith('.parquet'));

              if (hasParquetFiles) {
                const fileCount = fs
                  .readdirSync(fullPath)
                  .filter((file: string) => file.endsWith('.parquet')).length;
                app.debug(
                  `✅ Found SignalK path with data: ${newRelativePath} (${fileCount} files)`
                );
                paths.push({
                  path: newRelativePath,
                  directory: fullPath,
                  fileCount: fileCount,
                });
              } else {
                app.debug(
                  `📁 Directory ${newRelativePath} has no parquet files`
                );
              }

              walkPaths(fullPath, newRelativePath);
            }
          });
        } catch (error) {
          app.debug(
            `❌ Error reading directory ${currentPath}: ${(error as Error).message}`
          );
        }
      }

      if (fs.existsSync(vesselsDir)) {
        walkPaths(vesselsDir);
      }

      app.debug(
        `📊 Path discovery complete: found ${paths.length} paths with data`
      );
      return paths;
    }

    // Get available SignalK paths
    router.get(
      '/api/paths',
      (_: TypedRequest, res: TypedResponse<PathsApiResponse>) => {
        try {
          const dataDir = getDataDir();
          const paths = getAvailablePaths(dataDir);

          return res.json({
            success: true,
            dataDirectory: dataDir,
            paths: paths,
          });
        } catch (error) {
          return res.status(500).json({
            success: false,
            error: (error as Error).message,
          });
        }
      }
    );

    // Get files for a specific path
    router.get(
      '/api/files/:path(*)',
      (req: TypedRequest, res: TypedResponse<FilesApiResponse>) => {
        try {
          const dataDir = getDataDir();
          const signalkPath = req.params.path;
          const selfContextPath = app.selfContext
            .replace(/\./g, '/')
            .replace(/:/g, '_');
          const pathDir = path.join(
            dataDir,
            selfContextPath,
            signalkPath.replace(/\./g, '/')
          );

          if (!fs.existsSync(pathDir)) {
            return res.status(404).json({
              success: false,
              error: `Path not found: ${signalkPath}`,
            });
          }

          const files = fs
            .readdirSync(pathDir)
            .filter((file: string) => file.endsWith('.parquet'))
            .map((file: string) => {
              const filePath = path.join(pathDir, file);
              const stat = fs.statSync(filePath);
              return {
                name: file,
                path: filePath,
                size: stat.size,
                modified: stat.mtime.toISOString(),
              };
            })
            .sort(
              (a, b) =>
                new Date(b.modified).getTime() - new Date(a.modified).getTime()
            );

          return res.json({
            success: true,
            path: signalkPath,
            directory: pathDir,
            files: files,
          });
        } catch (error) {
          return res.status(500).json({
            success: false,
            error: (error as Error).message,
          });
        }
      }
    );

    // Get sample data from a specific file
    router.get(
      '/api/sample/:path(*)',
      async (req: TypedRequest, res: TypedResponse<SampleApiResponse>) => {
        try {
          const dataDir = getDataDir();
          const signalkPath = req.params.path;
          const limit = parseInt(req.query.limit as string) || 10;

          const selfContextPath = app.selfContext
            .replace(/\./g, '/')
            .replace(/:/g, '_');
          const pathDir = path.join(
            dataDir,
            selfContextPath,
            signalkPath.replace(/\./g, '/')
          );

          if (!fs.existsSync(pathDir)) {
            return res.status(404).json({
              success: false,
              error: `Path not found: ${signalkPath}`,
            });
          }

          // Get the most recent parquet file
          const files = fs
            .readdirSync(pathDir)
            .filter((file: string) => file.endsWith('.parquet'))
            .map((file: string) => {
              const filePath = path.join(pathDir, file);
              const stat = fs.statSync(filePath);
              return { name: file, path: filePath, modified: stat.mtime };
            })
            .sort((a, b) => b.modified.getTime() - a.modified.getTime());

          if (files.length === 0) {
            return res.status(404).json({
              success: false,
              error: `No parquet files found for path: ${signalkPath}`,
            });
          }

          const sampleFile = files[0];
          const query = `SELECT * FROM '${sampleFile.path}' LIMIT ${limit}`;

          const instance = await DuckDBInstance.create();
          const connection = await instance.connect();
          try {
            const reader = await connection.runAndReadAll(query);
            const rawData = reader.getRowObjects();

            const data = mapForJSON(rawData);

            // Get column info
            const columns = data.length > 0 ? Object.keys(data[0]) : [];

            return res.json({
              success: true,
              path: signalkPath,
              file: sampleFile.name,
              columns: columns,
              rowCount: data.length,
              data: data,
            });
          } catch (err) {
            return res.status(400).json({
              success: false,
              error: (err as Error).message,
            });
          } finally {
            connection.disconnectSync();
          }
        } catch (error) {
          return res.status(500).json({
            success: false,
            error: (error as Error).message,
          });
        }
      }
    );

    // Query parquet data
    router.post(
      '/api/query',
      async (
        req: TypedRequest<QueryRequest>,
        res: TypedResponse<QueryApiResponse>
      ) => {
        try {
          const { query } = req.body;

          if (!query) {
            return res.status(400).json({
              success: false,
              error: 'Query is required',
            });
          }

          const dataDir = getDataDir();

          // Replace placeholder paths in query with actual file paths
          let processedQuery = query;

          // Find all quoted paths in the query that might be SignalK paths
          const pathMatches = query.match(/'([^']+)'/g);
          if (pathMatches) {
            pathMatches.forEach(match => {
              const quotedPath = match.slice(1, -1); // Remove quotes

              // If it looks like a SignalK path, convert to file path
              const selfContextPath = toContextFilePath(
                app.selfContext as Context
              );
              if (
                quotedPath.includes(`/${selfContextPath}/`) ||
                quotedPath.includes('.parquet')
              ) {
                // It's already a file path, use as is
                return;
              } else if (
                quotedPath.includes('.') &&
                !quotedPath.includes('/')
              ) {
                // It's a SignalK path, convert to file path
                const filePath = toParquetFilePath(
                  dataDir,
                  selfContextPath,
                  quotedPath
                );
                processedQuery = processedQuery.replace(match, `'${filePath}'`);
              }
            });
          }

          app.debug(`Executing query: ${processedQuery}`);

          const instance = await DuckDBInstance.create();
          const connection = await instance.connect();
          try {
            const reader = await connection.runAndReadAll(processedQuery);
            const rawData = reader.getRowObjects();

            const data = mapForJSON(rawData);

            return res.json({
              success: true,
              query: processedQuery,
              rowCount: data.length,
              data: data,
            });
          } catch (err) {
            app.error(`Query error: ${err}`);
            return res.status(400).json({
              success: false,
              error: (err as Error).message,
            });
          } finally {
            connection.disconnectSync();
          }
        } catch (error) {
          return res.status(500).json({
            success: false,
            error: (error as Error).message,
          });
        }
      }
    );

    // Test S3 connection
    router.post(
      '/api/test-s3',
      async (_: TypedRequest, res: TypedResponse<S3TestApiResponse>) => {
        try {
          if (!state.currentConfig) {
            return res.status(500).json({
              success: false,
              error: 'Plugin not started or configuration not available',
            });
          }

          if (!state.currentConfig.s3Upload.enabled) {
            return res.status(400).json({
              success: false,
              error: 'S3 upload is not enabled in configuration',
            });
          }

          if (!S3Client || !state.s3Client) {
            return res.status(503).json({
              success: false,
              error: 'S3 client not available or not initialized',
            });
          }

          // Test S3 connection by listing bucket
          const listCommand = new ListObjectsV2Command({
            Bucket: state.currentConfig.s3Upload.bucket,
            MaxKeys: 1,
          });

          await state.s3Client.send(listCommand);

          return res.json({
            success: true,
            message: 'S3 connection successful',
            bucket: state.currentConfig.s3Upload.bucket,
            region: state.currentConfig.s3Upload.region || 'us-east-1',
            keyPrefix: state.currentConfig.s3Upload.keyPrefix || 'none',
          });
        } catch (error) {
          app.debug(`S3 test connection error: ${error}`);
          return res.status(500).json({
            success: false,
            error: (error as Error).message || 'S3 connection failed',
          });
        }
      }
    );

    // Web App Path Configuration API Routes (manages separate config file)

    // Get current path configurations
    router.get(
      '/api/config/paths',
      (_: TypedRequest, res: TypedResponse<ConfigApiResponse>) => {
        try {
          const paths = loadWebAppPaths();
          return res.json({
            success: true,
            paths: paths,
          });
        } catch (error) {
          return res.status(500).json({
            success: false,
            error: (error as Error).message,
          });
        }
      }
    );

    // Add new path configuration
    router.post(
      '/api/config/paths',
      (req: TypedRequest<PathConfigRequest>, res: TypedResponse): void => {
        try {
          const newPath = req.body;

          // Validate required fields
          if (!newPath.path) {
            res.status(400).json({
              success: false,
              error: 'Path is required',
            });
            return;
          }

          // Add to current paths
          currentPaths.push(newPath);

          // Save to web app configuration
          saveWebAppPaths(currentPaths);

          // Update subscriptions
          updateDataSubscriptions(state.currentConfig!);

          res.json({
            success: true,
            message: 'Path configuration added successfully',
            path: newPath,
          });
        } catch (error) {
          res.status(500).json({
            success: false,
            error: (error as Error).message,
          });
        }
      }
    );

    // Update existing path configuration
    router.put(
      '/api/config/paths/:index',
      (req: TypedRequest<PathConfigRequest>, res: TypedResponse): void => {
        try {
          const index = parseInt(req.params.index);
          const updatedPath = req.body;

          if (index < 0 || index >= currentPaths.length) {
            res.status(404).json({
              success: false,
              error: 'Path configuration not found',
            });
            return;
          }

          // Validate required fields
          if (!updatedPath.path) {
            res.status(400).json({
              success: false,
              error: 'Path is required',
            });
            return;
          }

          // Update the path configuration
          currentPaths[index] = updatedPath;

          // Save to web app configuration
          saveWebAppPaths(currentPaths);

          // Update subscriptions
          updateDataSubscriptions(state.currentConfig!);

          res.json({
            success: true,
            message: 'Path configuration updated successfully',
            path: updatedPath,
          });
        } catch (error) {
          res.status(500).json({
            success: false,
            error: (error as Error).message,
          });
        }
      }
    );

    // Remove path configuration
    router.delete(
      '/api/config/paths/:index',
      (req: TypedRequest, res: TypedResponse): void => {
        try {
          const index = parseInt(req.params.index);

          if (index < 0 || index >= currentPaths.length) {
            res.status(404).json({
              success: false,
              error: 'Path configuration not found',
            });
            return;
          }

          // Get the path being removed for response
          const removedPath = currentPaths[index];

          // Remove from current paths
          currentPaths.splice(index, 1);

          // Save to web app configuration
          saveWebAppPaths(currentPaths);

          // Update subscriptions
          updateDataSubscriptions(state.currentConfig!);

          app.debug(
            `Removed path configuration: ${removedPath.name} (${removedPath.path})`
          );

          res.json({
            success: true,
            message: 'Path configuration removed successfully',
            removedPath: removedPath,
          });
        } catch (error) {
          res.status(500).json({
            success: false,
            error: (error as Error).message,
          });
        }
      }
    );

    // Command Management API endpoints

    // Get all registered commands
    router.get(
      '/api/commands',
      (_: TypedRequest, res: TypedResponse<CommandApiResponse>) => {
        try {
          const commands = Array.from(commandState.registeredCommands.values());
          return res.json({
            success: true,
            commands: commands,
            count: commands.length,
          });
        } catch (error) {
          app.error(`Error retrieving commands: ${error}`);
          return res.status(500).json({
            success: false,
            error: 'Failed to retrieve commands',
          });
        }
      }
    );

    // Register a new command
    router.post(
      '/api/commands',
      (
        req: TypedRequest<CommandRegistrationRequest>,
        res: TypedResponse<CommandApiResponse>
      ) => {
        try {
          const { command, description } = req.body;

          if (!command || !isValidCommandName(command)) {
            return res.status(400).json({
              success: false,
              error:
                'Invalid command name. Must be alphanumeric with underscores, 1-50 characters.',
            });
          }

          const result = registerCommand(command, description);

          if (result.state === 'COMPLETED') {
            // Add to command history
            addCommandHistoryEntry(command, 'REGISTER', undefined, true);

            // Update webapp config
            const webAppConfig = loadWebAppConfig();
            saveWebAppConfig(webAppConfig.paths, currentCommands);

            const commandConfig = commandState.registeredCommands.get(command);
            return res.json({
              success: true,
              message: `Command '${command}' registered successfully`,
              command: commandConfig,
            });
          } else {
            return res.status(400).json({
              success: false,
              error: result.message || 'Failed to register command',
            });
          }
        } catch (error) {
          app.error(`Error registering command: ${error}`);
          return res.status(500).json({
            success: false,
            error: 'Internal server error',
          });
        }
      }
    );

    // Execute a command
    router.put(
      '/api/commands/:command/execute',
      (
        req: TypedRequest<CommandExecutionRequest>,
        res: TypedResponse<CommandExecutionResponse>
      ) => {
        try {
          const { command } = req.params;
          const { value } = req.body;

          if (typeof value !== 'boolean') {
            return res.status(400).json({
              success: false,
              error: 'Command value must be a boolean',
            });
          }

          const result = executeCommand(command, value);

          if (result.state === 'COMPLETED') {
            // Add to command history
            addCommandHistoryEntry(command, 'EXECUTE', value, true);

            return res.json({
              success: true,
              command: command,
              value: value,
              executed: true,
              timestamp: result.timestamp,
            });
          } else {
            // Add to command history for failed execution
            addCommandHistoryEntry(
              command,
              'EXECUTE',
              value,
              false,
              result.message
            );

            return res.status(400).json({
              success: false,
              error: result.message || 'Failed to execute command',
            });
          }
        } catch (error) {
          app.error(`Error executing command: ${error}`);
          return res.status(500).json({
            success: false,
            error: 'Internal server error',
          });
        }
      }
    );

    // Unregister a command
    router.delete(
      '/api/commands/:command',
      (req: TypedRequest, res: TypedResponse<CommandApiResponse>) => {
        try {
          const { command } = req.params;

          const result = unregisterCommand(command);

          if (result.state === 'COMPLETED') {
            // Add to command history
            addCommandHistoryEntry(command, 'UNREGISTER', undefined, true);

            // Update webapp config
            const webAppConfig = loadWebAppConfig();
            saveWebAppConfig(webAppConfig.paths, currentCommands);

            return res.json({
              success: true,
              message: `Command '${command}' unregistered successfully`,
            });
          } else {
            return res.status(404).json({
              success: false,
              error: result.message || 'Command not found',
            });
          }
        } catch (error) {
          app.error(`Error unregistering command: ${error}`);
          return res.status(500).json({
            success: false,
            error: 'Internal server error',
          });
        }
      }
    );

    // Get command history
    router.get(
      '/api/commands/history',
      (_: TypedRequest, res: TypedResponse<CommandApiResponse>) => {
        try {
          // Return the last 50 history entries
          const recentHistory = commandHistory.slice(-50);
          return res.json({
            success: true,
            data: recentHistory,
          });
        } catch (error) {
          app.error(`Error retrieving command history: ${error}`);
          return res.status(500).json({
            success: false,
            error: 'Failed to retrieve command history',
          });
        }
      }
    );

    // Get command status
    router.get(
      '/api/commands/:command/status',
      (req: TypedRequest, res: TypedResponse<CommandApiResponse>) => {
        try {
          const { command } = req.params;
          const commandConfig = commandState.registeredCommands.get(command);

          if (!commandConfig) {
            return res.status(404).json({
              success: false,
              error: 'Command not found',
            });
          }

          // Get current value from SignalK
          const currentValue = app.getSelfPath(`commands.${command}`);

          return res.json({
            success: true,
            command: {
              ...commandConfig,
              active: currentValue === true,
            },
          });
        } catch (error) {
          app.error(`Error retrieving command status: ${error}`);
          return res.status(500).json({
            success: false,
            error: 'Failed to retrieve command status',
          });
        }
      }
    );

    // Health check
    router.get(
      '/api/health',
      (_: TypedRequest, res: TypedResponse<HealthApiResponse>) => {
        return res.json({
          success: true,
          status: 'healthy',
          timestamp: new Date().toISOString(),
        });
      }
    );

    app.debug('Webapp API routes registered');
  };

  return plugin;
}

export function toContextFilePath(context: Context) {
  return context.replace(/\./g, '/').replace(/:/g, '_');
}

export function toParquetFilePath(
  dataDir: string,
  selfContextPath: string,
  quotedPath: string
) {
  return path.join(
    dataDir,
    selfContextPath,
    quotedPath.replace(/\./g, '/'),
    '*.parquet'
  );
}