import { Router, Request, Response } from 'express';
import {
  AggregateMethod,
  DataResult,
  FromToContextRequest,
  PathSpec,
} from './HistoryAPI-types';
import { ZonedDateTime, ZoneOffset, ZoneId } from '@js-joda/core';
import { Context, Path, Timestamp } from '@signalk/server-api';
import { ParamsDictionary } from 'express-serve-static-core';
import { ParsedQs } from 'qs';
import { DuckDBPool } from './utils/duckdb-pool';
import path from 'path';
import {
  getAvailablePathsArray,
  getAvailablePathsForTimeRange,
} from './utils/path-discovery';
import {
  getCachedPaths,
  setCachedPaths,
  getCachedContexts,
  setCachedContexts,
} from './utils/path-cache';
import {
  getAvailableContextsForTimeRange,
  getContextsInSpatialFilter,
} from './utils/context-discovery';
import { getPathComponentSchema, ComponentInfo } from './utils/schema-cache';
import { ConcurrencyLimiter } from './utils/concurrency-limiter';
import { CONCURRENCY } from './config/cache-defaults';
import { SQLiteBufferInterface } from './types';
import { HivePathBuilder, AggregationTier } from './utils/hive-path-builder';
import {
  AutoDiscoveryService,
  AutoDiscoveryResult,
} from './services/auto-discovery';
import {
  SpatialFilter,
  parseSpatialParams,
  buildSpatialSqlClause,
  isPositionPath,
} from './utils/spatial-queries';
import { calculateDistance } from './utils/geo-calculator';
import {
  buildBufferScalarSubquery,
  buildBufferObjectSubquery,
} from './utils/buffer-sql-builder';
import { stageBufferTable } from './utils/buffer-staging';
import {
  parseDurationToMillis,
  parseResolutionToMillis,
  InvalidResolutionError,
} from './utils/duration-parser';
import { isAngularPath } from './utils/angular-paths';
import { PathRetentionRule, RetentionRuleSet } from './utils/retention-rules';
import {
  parsePathFilters,
  buildParquetFilterClause,
  availableFilterColumns,
  filterEcho,
  FILTER_DELIMITERS,
} from './utils/path-filters';

// Allowed characters in the `paths` query parameter: path/aggregate tokens
// (alphanumerics, `.,:_`), the inline-filter delimiters from PATH_FILTER_DEFS,
// and `-`. Delimiters are escaped for use inside a character class; `-` is kept
// last so it stays a literal.
const PATHS_SANITIZER = new RegExp(
  `[^0-9a-z.,:_${FILTER_DELIMITERS.replace(/[\\\]^-]/g, '\\$&')}-]`,
  'gi'
);

export function registerHistoryApiRoute(
  router: Pick<Router, 'get'>,
  selfId: string,
  dataDir: string,
  debug: (k: string) => void,
  app: any,
  sqliteBuffer?: SQLiteBufferInterface,
  autoDiscoveryService?: AutoDiscoveryService,
  s3Config?: S3QueryConfig,
  pathRetentionOverrides?: PathRetentionRule[]
) {
  const historyApi = new HistoryAPI(
    selfId,
    dataDir,
    sqliteBuffer,
    autoDiscoveryService,
    s3Config,
    pathRetentionOverrides
  );
  // Handler for values endpoint
  const handleValues = (req: Request, res: Response) => {
    const { from, to, context, spatialFilter } = getRequestParams(
      req as FromToContextRequest,
      selfId
    );
    historyApi.getValues(
      context,
      from,
      to,
      spatialFilter,
      app,
      debug,
      req,
      res
    );
  };

  // V1 route: Direct routes with all signalk-parquet extensions
  // (spatial filtering, timezone conversion, moving averages, etc.)
  router.get('/signalk/v1/history/values', handleValues);
  // Handler for contexts endpoint
  const handleContexts = async (req: Request, res: Response) => {
    try {
      // Check if time range parameters are provided
      const hasTimeParams =
        req.query.duration || req.query.from || req.query.to;

      if (hasTimeParams) {
        // Time-range-aware: return only contexts with data in the specified time range
        const { from, to } = getRequestParams(
          req as FromToContextRequest,
          selfId
        );

        // Check cache first
        let contexts = getCachedContexts(from, to);

        if (!contexts) {
          // Cache miss - query the parquet files
          contexts = await getAvailableContextsForTimeRange(dataDir, from, to);
          // Cache the result
          setCachedContexts(from, to, contexts);
        }

        res.json(contexts);
      } else {
        // No time range specified: return only self context (legacy behavior)
        res.json([`vessels.${selfId}`] as Context[]);
      }
    } catch (error) {
      debug(`Error in /contexts: ${error}`);
      res.status(500).json({ error: (error as Error).message });
    }
  };

  // V1 route: Direct routes with extensions (time-range-aware context discovery)
  router.get('/signalk/v1/history/contexts', handleContexts);
  // Note: V2 routes (/signalk/v2/api/history/*) are handled by the registered
  // HistoryApi provider in history-provider.ts via app.registerHistoryApiProvider()
  // Handler for paths endpoint
  const handlePaths = async (req: Request, res: Response) => {
    try {
      // Check if time range parameters are provided
      const hasTimeParams =
        req.query.duration || req.query.from || req.query.to;

      if (hasTimeParams) {
        // Time-range-aware: return only paths with data in the specified time range
        const { from, to, context } = getRequestParams(
          req as FromToContextRequest,
          selfId
        );

        // Check cache first
        let paths = getCachedPaths(context, from, to);

        if (!paths) {
          // Cache miss - query the parquet files
          paths = await getAvailablePathsForTimeRange(
            dataDir,
            context,
            from,
            to
          );
          // Cache the result
          setCachedPaths(context, from, to, paths);
        }

        res.json(paths);
      } else {
        // No time range specified: return all available paths (legacy behavior)
        const context = req.query.context
          ? getContext(req.query.context as string, selfId)
          : undefined;
        const paths = getAvailablePathsArray(dataDir, app, context);
        res.json(paths);
      }
    } catch (error) {
      debug(`Error in /paths: ${error}`);
      res.status(500).json({ error: (error as Error).message });
    }
  };

  // V1 route: Direct routes with extensions (time-range-aware path discovery)
  router.get('/signalk/v1/history/paths', handlePaths);

  // Also register as plugin-style routes for testing
  router.get('/api/history/values', (req: Request, res: Response) => {
    const { from, to, context, spatialFilter } = getRequestParams(
      req as FromToContextRequest,
      selfId
    );
    historyApi.getValues(
      context,
      from,
      to,
      spatialFilter,
      app,
      debug,
      req,
      res
    );
  });
  router.get('/api/history/contexts', async (req: Request, res: Response) => {
    try {
      // Check if time range parameters are provided
      const hasTimeParams =
        req.query.duration || req.query.from || req.query.to;

      if (hasTimeParams) {
        // Time-range-aware: return only contexts with data in the specified time range
        const { from, to } = getRequestParams(
          req as FromToContextRequest,
          selfId
        );

        // Check cache first
        let contexts = getCachedContexts(from, to);

        if (!contexts) {
          // Cache miss - query the parquet files
          contexts = await getAvailableContextsForTimeRange(dataDir, from, to);
          // Cache the result
          setCachedContexts(from, to, contexts);
        }

        res.json(contexts);
      } else {
        // No time range specified: return only self context (legacy behavior)
        res.json([`vessels.${selfId}`] as Context[]);
      }
    } catch (error) {
      debug(`Error in /api/history/contexts: ${error}`);
      res.status(500).json({ error: (error as Error).message });
    }
  });
  router.get(
    '/api/history/contexts/spatial',
    async (req: Request, res: Response) => {
      try {
        const { from, to } = getRequestParams(
          req as FromToContextRequest,
          selfId
        );

        const spatialFilter = parseSpatialParams(
          req.query.bbox as string | undefined,
          req.query.radius as string | undefined
        );

        if (!spatialFilter) {
          res.status(400).json({
            error:
              'bbox (west,south,east,north) or radius (lon,lat,meters) is required',
          });
          return;
        }

        const contexts = await getContextsInSpatialFilter(
          dataDir,
          from,
          to,
          spatialFilter
        );
        res.json(contexts);
      } catch (error) {
        debug(`Error in /api/history/contexts/spatial: ${error}`);
        res.status(500).json({ error: (error as Error).message });
      }
    }
  );
  router.get('/api/history/paths', async (req: Request, res: Response) => {
    try {
      // Check if time range parameters are provided
      const hasTimeParams =
        req.query.duration || req.query.from || req.query.to;

      if (hasTimeParams) {
        // Time-range-aware: return only paths with data in the specified time range
        const { from, to, context } = getRequestParams(
          req as FromToContextRequest,
          selfId
        );

        // Check cache first
        let paths = getCachedPaths(context, from, to);

        if (!paths) {
          // Cache miss - query the parquet files
          paths = await getAvailablePathsForTimeRange(
            dataDir,
            context,
            from,
            to
          );
          // Cache the result
          setCachedPaths(context, from, to, paths);
        }

        res.json(paths);
      } else {
        // No time range specified: return all available paths (legacy behavior)
        const context = req.query.context
          ? getContext(req.query.context as string, selfId)
          : undefined;
        const paths = getAvailablePathsArray(dataDir, app, context);
        res.json(paths);
      }
    } catch (error) {
      debug(`Error in /api/history/paths: ${error}`);
      res.status(500).json({ error: (error as Error).message });
    }
  });
}

const getRequestParams = ({ query }: FromToContextRequest, selfId: string) => {
  try {
    let from: ZonedDateTime;
    let to: ZonedDateTime;

    // ============================================================================
    // STANDARD SIGNALK TIME RANGE PATTERNS
    // ============================================================================
    // Pattern 1: duration only → query back from now
    if (query.duration && !query.from && !query.to) {
      const durationMs = parseDuration(query.duration);
      to = ZonedDateTime.now(ZoneOffset.UTC);
      from = to.minusNanos(durationMs * 1000000);
    }
    // Pattern 2: from + duration → query forward from start
    else if (query.from && query.duration && !query.to) {
      from = parseDateTime(query.from);
      const durationMs = parseDuration(query.duration);
      to = from.plusNanos(durationMs * 1000000);
    }
    // Pattern 3: to + duration → query backward to end
    else if (query.to && query.duration && !query.from) {
      to = parseDateTime(query.to);
      const durationMs = parseDuration(query.duration);
      from = to.minusNanos(durationMs * 1000000);
    }
    // Pattern 4: from only → from start to now
    else if (query.from && !query.duration && !query.to) {
      from = parseDateTime(query.from);
      to = ZonedDateTime.now(ZoneOffset.UTC);
    }
    // Pattern 5: from + to → specific range
    else if (query.from && query.to && !query.duration) {
      from = parseDateTime(query.from);
      to = parseDateTime(query.to);
    } else {
      throw new Error(
        'Invalid time range parameters. Use one of the following patterns:\n' +
          '  1. ?duration=1h (query back from now)\n' +
          '  2. ?from=2025-08-01T00:00:00Z&duration=1h (query forward)\n' +
          '  3. ?to=2025-08-01T12:00:00Z&duration=1h (query backward)\n' +
          '  4. ?from=2025-08-01T00:00:00Z (from start to now)\n' +
          '  5. ?from=2025-08-01T00:00:00Z&to=2025-08-02T00:00:00Z (specific range)'
      );
    }

    const context: Context = getContext(query.context, selfId);
    const spatialFilter = parseSpatialParams(query.bbox, query.radius);
    return { from, to, context, spatialFilter };
  } catch (e: unknown) {
    console.error('Full error details:', e);
    throw new Error(
      `Error extracting query parameters from ${JSON.stringify(query)}: ${e instanceof Error ? e.stack : e}`
    );
  }
};

// Parse duration string (supports ISO 8601, integer seconds, shorthand)
function parseDuration(duration: string | undefined): number {
  if (!duration) {
    throw new Error('Duration parameter is required');
  }
  return parseDurationToMillis(duration);
}

// Check if datetime string has timezone information
function hasTimezoneInfo(dateTimeStr: string): boolean {
  // Check for 'Z' at the end, or '+'/'-' followed by timezone offset pattern
  return (
    dateTimeStr.endsWith('Z') ||
    /[+-]\d{2}:?\d{2}$/.test(dateTimeStr) ||
    /[+-]\d{4}$/.test(dateTimeStr)
  );
}

// Parse datetime string per ISO 8601:
// - Bare timestamps (no Z, no offset) → local time, converted to UTC
// - Timestamps with Z or offset → parsed as-is
function parseDateTime(dateTimeStr: string): ZonedDateTime {
  // Normalize the datetime string to include seconds if missing
  let normalizedStr = dateTimeStr;
  if (dateTimeStr.match(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}$/)) {
    // Add seconds if only HH:MM is provided
    normalizedStr = dateTimeStr + ':00';
  }

  if (hasTimezoneInfo(normalizedStr)) {
    // Has timezone info (Z or offset), parse as-is and convert to UTC
    return ZonedDateTime.parse(normalizedStr).withZoneSameInstant(
      ZoneOffset.UTC
    );
  } else {
    // No timezone info — per ISO 8601, treat as local time and convert to UTC
    try {
      // JavaScript Date constructor treats ISO strings without timezone as local time
      const localDate = new Date(normalizedStr);
      if (isNaN(localDate.getTime())) {
        throw new Error('Invalid date');
      }

      // Convert to UTC ISO string and parse with ZonedDateTime
      const utcIsoString = localDate.toISOString();
      return ZonedDateTime.parse(utcIsoString);
    } catch (e) {
      throw new Error(
        `Unable to parse datetime '${dateTimeStr}': ${e}. Use format like '2025-08-13T08:00:00' or '2025-08-13T08:00:00Z'`
      );
    }
  }
}

function getContext(
  contextFromQuery: string | undefined,
  selfId: string
): Context {
  if (
    !contextFromQuery ||
    contextFromQuery === 'vessels.self' ||
    contextFromQuery === 'self'
  ) {
    return `vessels.${selfId}` as Context;
  }
  return contextFromQuery.replace(/ /gi, '') as Context;
}

type QuerySource = 'local' | 's3' | 'hybrid' | 'auto';

export interface S3QueryConfig {
  enabled: boolean;
  bucket: string;
  keyPrefix: string;
  region: string;
}

/**
 * Stream a DataResult to the response as a single JSON document,
 * progressively, instead of buffering it via res.json(). Output is
 * byte-equivalent to the prior res.json(result) path: same field
 * order (context, range, values, data, units?, meta?), same UTC->
 * local timestamp conversion on the range and on each row[0].
 *
 * Wrapper construction: build the wrapper object with a `data: []`
 * placeholder, JSON.stringify it once, then split at that placeholder.
 * The two halves frame the row-by-row writes. This avoids hand-rolling
 * `,` / `{}` / `[]` framing tokens.
 *   Input wrapper -> '{...,"data":[],"units":{...}}'
 *   prefix -> '{...,"data":['
 *   suffix -> ']' + ',"units":{...}}'
 *
 * Backpressure: if res.write returns false the loop awaits 'drain'
 * before continuing, so a slow client cannot grow Node's internal
 * write buffer beyond the high-water mark. The wait races against
 * 'close' so a client that disconnects mid-flight does not hang the
 * stream forever.
 *
 * Cancellation: a 'close' listener on res sets a flag so the loop
 * breaks early on client disconnect. The DuckDB result has already
 * been materialised before this function is called, so cancellation
 * is purely about cutting the write loop short.
 *
 * Logging: on success or post-headers failure, emits one structured
 * line via the optional `log` callback with row count, byte count
 * and wall-clock duration. Operators chasing "endpoint feels slow"
 * have a single line to grep for.
 */
async function streamDataResult(
  res: Response,
  result: DataResult,
  log?: (msg: string) => void
): Promise<void> {
  const startMs = Date.now();
  let rowsWritten = 0;
  let bytesWritten = 0;
  let clientGone = false;
  const onClose = (): void => {
    clientGone = true;
  };
  // A mid-stream socket failure (ECONNRESET / EPIPE) emits 'error' on res.
  // With no listener Node rethrows it as an uncaughtException and takes down
  // the whole server process. Treat it like a disconnect: flip the flag so the
  // write loop unwinds instead of writing into a broken socket.
  const onError = (): void => {
    clientGone = true;
  };
  res.on('close', onClose);
  res.on('error', onError);

  try {
    res.setHeader('Content-Type', 'application/json; charset=utf-8');
    // Streaming responses must not be cached by an intermediary.
    // X-Accel-Buffering is nginx-specific (no-op elsewhere) and tells
    // nginx not to buffer the response body, so chunks reach the
    // client as we write them rather than after res.end().
    res.setHeader('Cache-Control', 'no-store');
    res.setHeader('X-Accel-Buffering', 'no');

    const localRange = {
      from: utcToLocalTimestamp(result.range.from),
      to: utcToLocalTimestamp(result.range.to),
    };

    // Build the wrapper with `data: []` placeholder so JSON.stringify
    // produces the framing for us. Optional fields (units, meta) come
    // after data per DataResult declaration order; insertion order in
    // a plain object is preserved by JSON.stringify.
    const wrapper: Record<string, unknown> = {
      context: result.context,
      range: localRange,
      values: result.values,
      data: [],
    };
    if (result.units) wrapper.units = result.units;
    if (result.meta) wrapper.meta = result.meta;

    const wrapperJson = JSON.stringify(wrapper);
    const marker = '"data":[]';
    const markerIdx = wrapperJson.indexOf(marker);
    if (markerIdx === -1) {
      // The wrapper is constructed locally with data:[] as a literal
      // field; this branch only fires if a future JSON.stringify ever
      // omits empty arrays. Cheaper to assert than to debug a silent
      // wire-format break.
      throw new Error('streamDataResult: wrapper marker not found');
    }
    const prefix = wrapperJson.slice(0, markerIdx) + '"data":[';
    const suffix = ']' + wrapperJson.slice(markerIdx + marker.length);

    const writeOrAwait = async (chunk: string): Promise<void> => {
      bytesWritten += Buffer.byteLength(chunk, 'utf8');
      if (!res.write(chunk)) {
        // Wait for drain (kernel buffer flushed), close (client gone), or
        // error (socket broke). Whichever fires first wins; all three unhook
        // the others so we don't leak handlers across the loop, and we never
        // await a 'drain' that a dead socket will never emit.
        await new Promise<void>(resolve => {
          let settled = false;
          const finish = (): void => {
            if (settled) return;
            settled = true;
            res.off('drain', finish);
            res.off('close', finish);
            res.off('error', finish);
            resolve();
          };
          res.once('drain', finish);
          res.once('close', finish);
          res.once('error', finish);
        });
      }
    };

    await writeOrAwait(prefix);

    const data = result.data;
    for (let i = 0; i < data.length; i++) {
      if (clientGone) break;
      const row = data[i];
      // Allocate a converted row instead of mutating in place. The
      // input is logically owned by the caller, and a future second
      // caller (audit log, dual-format response) would silently see
      // local time where it expected UTC.
      const converted = row.slice() as typeof row;
      converted[0] = utcToLocalTimestamp(row[0]);
      const rowChunk = (i > 0 ? ',' : '') + JSON.stringify(converted);
      await writeOrAwait(rowChunk);
      rowsWritten++;
    }

    if (clientGone) {
      log?.(
        `streamDataResult: client disconnected after ${rowsWritten}/${data.length} rows, ${bytesWritten} bytes, ${Date.now() - startMs}ms`
      );
      return;
    }

    await writeOrAwait(suffix);
    res.end();
    log?.(
      `streamDataResult: rows=${rowsWritten} bytes=${bytesWritten} duration=${Date.now() - startMs}ms`
    );
  } finally {
    res.off('close', onClose);
    res.off('error', onError);
  }
}

/**
 * Buffered UTC->local conversion of a full DataResult. Used by callers
 * that pass a non-streaming response (e.g. internal mocks that only
 * implement res.json) so they receive the same locally-converted shape
 * the streaming path produces.
 */
function convertToLocalTime(result: DataResult): DataResult {
  return {
    ...result,
    range: {
      from: utcToLocalTimestamp(result.range.from),
      to: utcToLocalTimestamp(result.range.to),
    },
    data: result.data.map(row => {
      const newRow = row.slice() as typeof row;
      newRow[0] = utcToLocalTimestamp(row[0]);
      return newRow;
    }),
  };
}

/**
 * Returns true when `res` exposes the Node stream methods streamDataResult
 * relies on. The plugin's HTTP routes pass a real Express Response, but
 * historical-streaming.ts reuses getValues with a mock that only implements
 * res.json / res.status().json — those callers go through the buffered path.
 */
function canStreamResponse(res: Response): boolean {
  return (
    typeof (res as Partial<Response>).write === 'function' &&
    typeof (res as Partial<Response>).end === 'function' &&
    typeof (res as Partial<Response>).setHeader === 'function' &&
    typeof (res as Partial<Response>).on === 'function'
  );
}

/**
 * Convert a UTC timestamp string to server-local time with offset suffix.
 * e.g. "2026-03-06T19:00:00Z" → "2026-03-06T14:00:00-05:00" (EST)
 */
function utcToLocalTimestamp(utcTs: Timestamp): Timestamp {
  try {
    const zdt = ZonedDateTime.parse(utcTs).withZoneSameInstant(
      ZoneId.systemDefault()
    );
    // ZonedDateTime.toString() appends "[SYSTEM]" zone ID — strip it
    return zdt.toString().replace(/\[.*\]$/, '') as Timestamp;
  } catch {
    // If parsing fails (e.g. already has offset or unusual format), try via Date
    const d = new Date(utcTs);
    if (isNaN(d.getTime())) return utcTs;
    const offsetMin = d.getTimezoneOffset();
    const local = new Date(d.getTime() - offsetMin * 60000);
    const sign = offsetMin <= 0 ? '+' : '-';
    const absH = String(Math.floor(Math.abs(offsetMin) / 60)).padStart(2, '0');
    const absM = String(Math.abs(offsetMin) % 60).padStart(2, '0');
    return (local.toISOString().replace('Z', '') +
      sign +
      absH +
      ':' +
      absM) as Timestamp;
  }
}

// A response `values` entry: path + aggregate method, optional smoothing
// metadata, plus any echoed inline-filter fields (e.g. `sourceRef`). The index
// signature keeps it open to new filters without a type change here.
interface HistoryValueEntry {
  path: Path;
  method: AggregateMethod;
  smoothing?: string;
  window?: number;
  [field: string]: unknown;
}

export class HistoryAPI {
  private sqliteBuffer?: SQLiteBufferInterface;
  private hivePathBuilder: HivePathBuilder;
  private autoDiscoveryService?: AutoDiscoveryService;
  private s3Config?: S3QueryConfig;
  // Mirrors AggregationService's rule set so the read path can fall
  // back to tier=raw for skipAggregation paths. The aggregation
  // pipeline doesn't write 5s/60s/1h files for those, so a higher-
  // resolution query against an aggregated tier would otherwise return
  // empty.
  private retentionRules: RetentionRuleSet;

  constructor(
    private selfId: string,
    private dataDir: string,
    sqliteBuffer?: SQLiteBufferInterface,
    autoDiscoveryService?: AutoDiscoveryService,
    s3Config?: S3QueryConfig,
    pathRetentionOverrides?: PathRetentionRule[]
  ) {
    this.sqliteBuffer = sqliteBuffer;
    this.hivePathBuilder = new HivePathBuilder();
    this.autoDiscoveryService = autoDiscoveryService;
    this.s3Config = s3Config;
    this.retentionRules = new RetentionRuleSet(pathRetentionOverrides || []);
  }

  /**
   * Set S3 configuration for federated queries
   */
  setS3Config(config: S3QueryConfig | undefined): void {
    this.s3Config = config;
  }

  /**
   * Set the auto-discovery service
   */
  setAutoDiscoveryService(service: AutoDiscoveryService | undefined): void {
    this.autoDiscoveryService = service;
  }

  /**
   * Set the SQLite buffer for federated queries
   */
  setSqliteBuffer(buffer: SQLiteBufferInterface | undefined): void {
    this.sqliteBuffer = buffer;
  }

  /**
   * Auto-select the optimal tier based on requested resolution
   * Returns undefined to use raw/flat data, or a tier name for aggregated data
   *
   * Logic:
   * - resolution >= 1 hour (3600000ms) → use 1h tier
   * - resolution >= 1 minute (60000ms) → use 60s tier
   * - resolution >= 5 seconds (5000ms) → use 5s tier
   * - resolution < 5 seconds → use raw data (no tier)
   *
   * Falls back through tiers if preferred tier doesn't exist
   */
  private selectOptimalTier(
    resolutionMillis: number
  ): AggregationTier | undefined {
    // eslint-disable-next-line @typescript-eslint/no-require-imports
    const fs = require('fs');

    // Determine preferred tier based on resolution
    let preferredTiers: AggregationTier[] = [];

    if (resolutionMillis >= 3600000) {
      preferredTiers = ['1h', '60s', '5s'];
    } else if (resolutionMillis >= 60000) {
      preferredTiers = ['60s', '5s'];
    } else if (resolutionMillis >= 5000) {
      preferredTiers = ['5s'];
    } else {
      // Use raw data for sub-5-second resolution
      return undefined;
    }

    // Check which tiers exist and return the best available
    for (const tier of preferredTiers) {
      const tierPath = path.join(this.dataDir, `tier=${tier}`);
      try {
        if (fs.existsSync(tierPath)) {
          return tier;
        }
      } catch {
        // Directory doesn't exist or not accessible
      }
    }

    // No aggregated tiers available, use raw data
    return undefined;
  }

  /**
   * Get timestamps where vessel position was within the spatial filter
   * Used to correlate non-position paths with spatial filtering
   */
  private async getSpatialTimestamps(
    context: Context,
    from: ZonedDateTime,
    to: ZonedDateTime,
    timeResolutionMillis: number,
    spatialFilter: SpatialFilter,
    positionPath: string,
    _tier: AggregationTier | undefined,
    _querySource: QuerySource,
    debug: (k: string) => void
  ): Promise<Set<string>> {
    const timestamps = new Set<string>();

    // Build file path for raw position data
    const sanitizedContext = this.hivePathBuilder.sanitizeContext(context);
    const sanitizedPath = this.hivePathBuilder.sanitizePath(positionPath);
    const localFilePath = path.join(
      this.dataDir,
      'tier=raw',
      `context=${sanitizedContext}`,
      `path=${sanitizedPath}`,
      '**',
      '*.parquet'
    );

    const fromIso = from.toInstant().toString();
    const toIso = to.toInstant().toString();

    try {
      const hasBuffer = DuckDBPool.isSQLiteBufferInitialized();
      const connection = await DuckDBPool.getConnection();
      try {
        // Bucket-lookup approach: instead of scanning all raw position data,
        // bucket by time resolution, grab FIRST lat/lon per bucket, then filter by bbox/radius.
        // This reads far less data than a full scan with spatial SQL.
        const bucketExpr = `strftime(DATE_TRUNC('seconds',
          EPOCH_MS(CAST(FLOOR(EPOCH_MS(signalk_timestamp::TIMESTAMP) / ${timeResolutionMillis}) * ${timeResolutionMillis} AS BIGINT))
        ), '%Y-%m-%dT%H:%M:%SZ')`;

        // Build FROM: parquet UNION ALL buffer
        const parquetFrom = `SELECT signalk_timestamp, value_latitude, value_longitude FROM (
          SELECT * FROM read_parquet('${localFilePath}', union_by_name=true, filename=true)
          WHERE filename NOT LIKE '%/processed/%'
          AND filename NOT LIKE '%/quarantine/%'
          AND filename NOT LIKE '%/failed/%'
          AND filename NOT LIKE '%/repaired/%')`;

        let fromSource = `(${parquetFrom})`;
        if (hasBuffer && this.sqliteBuffer) {
          const bufferTableCols =
            this.sqliteBuffer.getTableColumns(positionPath);
          const stagedTable = await stageBufferTable(
            connection,
            this.sqliteBuffer,
            String(context),
            positionPath,
            fromIso,
            toIso,
            debug
          );
          const bufferSubquery = stagedTable
            ? buildBufferObjectSubquery(
                stagedTable,
                context,
                fromIso,
                toIso,
                new Map([
                  [
                    'latitude',
                    {
                      name: 'latitude',
                      columnName: 'value_latitude',
                      dataType: 'numeric' as const,
                    },
                  ],
                  [
                    'longitude',
                    {
                      name: 'longitude',
                      columnName: 'value_longitude',
                      dataType: 'numeric' as const,
                    },
                  ],
                ]),
                bufferTableCols
              )
            : null;
          if (bufferSubquery) {
            fromSource = `(${parquetFrom} UNION ALL SELECT signalk_timestamp, TRY_CAST(value_latitude AS DOUBLE) as value_latitude, TRY_CAST(value_longitude AS DOUBLE) as value_longitude FROM ${bufferSubquery})`;
          }
        }

        const query = `
          SELECT
            ${bucketExpr} as timestamp,
            FIRST(TRY_CAST(value_latitude AS DOUBLE)) as lat,
            FIRST(TRY_CAST(value_longitude AS DOUBLE)) as lon
          FROM ${fromSource}
          WHERE
            signalk_timestamp >= '${fromIso}'
            AND signalk_timestamp < '${toIso}'
            AND TRY_CAST(value_latitude AS DOUBLE) IS NOT NULL
            AND TRY_CAST(value_longitude AS DOUBLE) IS NOT NULL
          GROUP BY timestamp
          ORDER BY timestamp
        `;

        const result = await connection.runAndReadAll(query);
        const rows = result.getRowObjects() as Array<{
          timestamp: string;
          lat: number;
          lon: number;
        }>;

        // Filter buckets by spatial filter in JS
        const { bbox } = spatialFilter;
        for (const row of rows) {
          if (row.lat === null || row.lon === null) continue;

          // Lat check
          if (row.lat < bbox.south || row.lat > bbox.north) continue;

          // Lon check (handles 180° meridian crossing)
          if (bbox.west <= bbox.east) {
            if (row.lon < bbox.west || row.lon > bbox.east) continue;
          } else {
            if (row.lon < bbox.west && row.lon > bbox.east) continue;
          }

          // Precise radius check if applicable
          if (
            spatialFilter.type === 'radius' &&
            spatialFilter.centerLat !== undefined &&
            spatialFilter.centerLon !== undefined &&
            spatialFilter.radiusMeters !== undefined
          ) {
            const dist = calculateDistance(
              row.lat,
              row.lon,
              spatialFilter.centerLat,
              spatialFilter.centerLon
            );
            if (dist > spatialFilter.radiusMeters) continue;
          }

          timestamps.add(row.timestamp);
        }

        debug(
          `[Spatial Correlation] Bucketed ${rows.length} positions, ${timestamps.size} within spatial filter`
        );
      } finally {
        connection.disconnectSync();
      }
    } catch (error) {
      debug(`[Spatial Correlation] Error querying position data: ${error}`);
      // On error, return empty set (no filtering will occur)
    }

    return timestamps;
  }

  async getValues(
    context: Context,
    from: ZonedDateTime,
    to: ZonedDateTime,
    spatialFilter: SpatialFilter | null,
    app: any,
    debug: (k: string) => void,
    // eslint-disable-next-line @typescript-eslint/no-explicit-any
    req: Request<ParamsDictionary, any, any, ParsedQs, Record<string, any>>,
    // eslint-disable-next-line @typescript-eslint/no-explicit-any
    res: Response<any, Record<string, any>>
  ) {
    try {
      // Resolution now in SECONDS (breaking change from v0.7.0)
      const timeResolutionMillis = req.query.resolution
        ? parseResolutionToMillis(req.query.resolution as string)
        : ((to.toEpochSecond() - from.toEpochSecond()) / 500) * 1000;
      // Strip everything except path/aggregate characters, the filter
      // delimiters from PATH_FILTER_DEFS, and `-` (common in source
      // references). This is a coarse injection guard; values are also SQL-
      // escaped downstream. Adding a delimiter to the registry extends this
      // automatically.
      const pathExpressions = ((req.query.paths as string) || '')
        .replace(PATHS_SANITIZER, '')
        .split(',');
      const pathSpecs: PathSpec[] = pathExpressions.map(splitPathExpression);

      // Auto-select tier based on resolution (provider selects automatically)
      const tier = this.selectOptimalTier(timeResolutionMillis);
      if (tier) {
        debug(
          `Auto-selected tier=${tier} for resolution=${timeResolutionMillis}ms`
        );
      }

      // Log spatial filter if present
      if (spatialFilter) {
        debug(
          `Spatial filter: type=${spatialFilter.type}, bbox=[${spatialFilter.bbox.west},${spatialFilter.bbox.south},${spatialFilter.bbox.east},${spatialFilter.bbox.north}]`
        );
      }

      // Handle position and numeric paths together
      const positionPath = 'navigation.position';
      const allResult = pathSpecs.length
        ? await this.getNumericValues(
            context,
            from,
            to,
            timeResolutionMillis,
            pathSpecs,
            debug,
            tier,
            spatialFilter,
            'local',
            positionPath,
            app
          )
        : {
            context,
            range: {
              from: from.toString() as Timestamp,
              to: to.toString() as Timestamp,
            },
            values: [],
            data: [],
          };

      // Check for auto-discovery on paths with no data
      debug(
        `[AutoDiscovery] Checking auto-discovery: service=${!!this.autoDiscoveryService}, pathSpecs.length=${pathSpecs.length}`
      );
      if (this.autoDiscoveryService && pathSpecs.length > 0) {
        const autoConfiguredPaths: AutoDiscoveryResult[] = [];

        for (let i = 0; i < pathSpecs.length; i++) {
          const pathSpec = pathSpecs[i];
          // Check if this path has any data in the result
          // Data array format: [timestamp, value1, value2, ...]
          // Each path corresponds to a position in the values array
          const hasData = allResult.data.some(row => {
            const valueIndex = i + 1; // +1 because index 0 is timestamp
            return row[valueIndex] !== null && row[valueIndex] !== undefined;
          });

          if (!hasData) {
            debug(
              `[AutoDiscovery] No data found for path ${pathSpec.path}, checking auto-discovery`
            );
            const result =
              await this.autoDiscoveryService.maybeAutoConfigurePath(
                pathSpec.path as Path,
                context
              );
            if (result.configured) {
              autoConfiguredPaths.push(result);
              debug(`[AutoDiscovery] Auto-configured path: ${pathSpec.path}`);
            } else {
              debug(
                `[AutoDiscovery] Path ${pathSpec.path} not auto-configured: ${result.reason}`
              );
            }
          }
        }

        // Add meta to response if any paths were auto-configured
        if (autoConfiguredPaths.length > 0) {
          allResult.meta = {
            autoConfigured: true,
            paths: autoConfiguredPaths.map(r => r.path),
            message: `${autoConfiguredPaths.length} path(s) auto-configured for recording. Data will be available shortly.`,
          };
        }
      }

      if (canStreamResponse(res)) {
        await streamDataResult(res, allResult, debug);
      } else {
        // Internal callers (historical-streaming.ts) pass a mock res with
        // only json()/status().json(); fall back to the buffered shape so
        // they keep working without an Express Response.
        res.json(convertToLocalTime(allResult));
      }
    } catch (error) {
      if (error instanceof InvalidResolutionError) {
        debug(`Bad request in getValues: ${error.message}`);
        res.status(400).json({ error: error.message });
        return;
      }
      // Note: the response may have been streamed, so by the time an
      // error reaches here, headers may already be on the wire. The two
      // branches below are deliberately not consolidated into a single
      // res.status(500) call as in the rest of this module.
      if (res.headersSent) {
        // Stream already started -- can't switch to a 500. Destroy the socket
        // rather than res.end(): ending would close a half-written JSON array
        // cleanly, so the client would see an orderly 200 indistinguishable
        // from a complete body. Destroying surfaces a broken transfer instead.
        // The message is distinct from the pre-stream failure path so an
        // operator can tell a truncated mid-stream failure apart from it.
        debug(
          `Error in getValues after headersSent (truncated response): ${error}`
        );
        res.destroy();
        return;
      }
      debug(`Error in getValues: ${error}`);
      res.status(500).json({
        error: 'Internal server error',
        message: error instanceof Error ? error.message : String(error),
      });
    }
  }

  async getNumericValues(
    context: Context,
    from: ZonedDateTime,
    to: ZonedDateTime,
    timeResolutionMillis: number,
    pathSpecs: PathSpec[],
    debug: (k: string) => void,
    tier?: AggregationTier,
    spatialFilter?: SpatialFilter | null,
    querySource: QuerySource = 'local',
    positionPath: string = 'navigation.position',
    app?: any
  ): Promise<DataResult> {
    // Keyed by pathSpecKey(spec), not bare path, so the same path requested
    // with different sources/aggregates keeps separate series.
    const allData: { [key: string]: Array<[Timestamp, unknown]> } = {};
    const objectPaths = new Set<string>(); // Track which paths are object paths

    // Convert ZonedDateTime to Date for S3 pattern building
    const fromDate = new Date(from.toInstant().toString());
    const toDate = new Date(to.toInstant().toString());

    // If spatial filter is set, get valid timestamps from position data
    // This allows filtering non-position paths by "when vessel was in this area"
    let spatialTimestamps: Set<string> | null = null;
    const hasPositionPath =
      spatialFilter && pathSpecs.some(ps => isPositionPath(ps.path));
    if (spatialFilter) {
      const hasNonPositionPaths = pathSpecs.some(
        ps => !isPositionPath(ps.path)
      );
      if (hasNonPositionPaths && !hasPositionPath) {
        // Position not in requested paths — need a separate scan for correlation
        debug(
          `[Spatial Correlation] Querying ${positionPath} for valid timestamps within spatial filter`
        );
        spatialTimestamps = await this.getSpatialTimestamps(
          context,
          from,
          to,
          timeResolutionMillis,
          spatialFilter,
          positionPath,
          undefined, // Force raw tier — position is an object path, aggregated tiers lack value_latitude/value_longitude
          querySource,
          debug
        );
        debug(
          `[Spatial Correlation] Found ${spatialTimestamps.size} valid timestamps`
        );
      }

      if (hasPositionPath) {
        // Position IS in requested paths — use fast bucket+FIRST query for both
        // position results AND spatial correlation timestamps (one query, no full raw scan)
        const posPathSpec = pathSpecs.find(ps => isPositionPath(ps.path))!;
        debug(
          `[Spatial] Fast bucket query for ${posPathSpec.path} (FIRST lat/lon per bucket + JS filter)`
        );

        const sanitizedCtx = this.hivePathBuilder.sanitizeContext(context);
        const sanitizedPos = this.hivePathBuilder.sanitizePath(
          posPathSpec.path
        );
        const posFilePath = path.join(
          this.dataDir,
          'tier=raw',
          `context=${sanitizedCtx}`,
          `path=${sanitizedPos}`,
          '**',
          '*.parquet'
        );
        const fromIso = from.toInstant().toString();
        const toIso = to.toInstant().toString();

        try {
          const hasBuffer = DuckDBPool.isSQLiteBufferInitialized();
          const connection = await DuckDBPool.getConnection();
          try {
            const bucketExpr = `strftime(DATE_TRUNC('seconds',
              EPOCH_MS(CAST(FLOOR(EPOCH_MS(signalk_timestamp::TIMESTAMP) / ${timeResolutionMillis}) * ${timeResolutionMillis} AS BIGINT))
            ), '%Y-%m-%dT%H:%M:%SZ')`;

            // Optional filters for the position path (e.g. one of several GPS
            // receivers). This fast path is local-only (raw parquet + buffer),
            // so probe just the local glob for the filter columns.
            const posAvailable = await availableFilterColumns(
              connection,
              [posFilePath],
              posPathSpec.filters
            );
            const posSourceFilter = buildParquetFilterClause(
              posPathSpec.filters,
              posAvailable
            );

            // Build FROM: parquet UNION ALL buffer (for today's unexported data)
            const parquetFrom = `SELECT signalk_timestamp, value_latitude, value_longitude FROM (
              SELECT * FROM read_parquet('${posFilePath}', union_by_name=true, filename=true)
              WHERE filename NOT LIKE '%/processed/%'
              AND filename NOT LIKE '%/quarantine/%'
              AND filename NOT LIKE '%/failed/%'
              AND filename NOT LIKE '%/repaired/%'${posSourceFilter})`;

            let fromSource = `(${parquetFrom})`;
            if (hasBuffer && this.sqliteBuffer) {
              const bufferTableCols = this.sqliteBuffer.getTableColumns(
                posPathSpec.path
              );
              const stagedTable = await stageBufferTable(
                connection,
                this.sqliteBuffer,
                String(context),
                posPathSpec.path,
                fromIso,
                toIso,
                debug
              );
              const bufferSubquery = stagedTable
                ? buildBufferObjectSubquery(
                    stagedTable,
                    context,
                    fromIso,
                    toIso,
                    new Map([
                      [
                        'latitude',
                        {
                          name: 'latitude',
                          columnName: 'value_latitude',
                          dataType: 'numeric' as const,
                        },
                      ],
                      [
                        'longitude',
                        {
                          name: 'longitude',
                          columnName: 'value_longitude',
                          dataType: 'numeric' as const,
                        },
                      ],
                    ]),
                    bufferTableCols,
                    posPathSpec.filters
                  )
                : null;
              if (bufferSubquery) {
                fromSource = `(${parquetFrom} UNION ALL SELECT signalk_timestamp, TRY_CAST(value_latitude AS DOUBLE) as value_latitude, TRY_CAST(value_longitude AS DOUBLE) as value_longitude FROM ${bufferSubquery})`;
              }
            }

            const query = `
              SELECT
                ${bucketExpr} as timestamp,
                FIRST(TRY_CAST(value_latitude AS DOUBLE)) as lat,
                FIRST(TRY_CAST(value_longitude AS DOUBLE)) as lon
              FROM ${fromSource}
              WHERE
                signalk_timestamp >= '${fromIso}'
                AND signalk_timestamp < '${toIso}'
                AND TRY_CAST(value_latitude AS DOUBLE) IS NOT NULL
                AND TRY_CAST(value_longitude AS DOUBLE) IS NOT NULL
              GROUP BY timestamp
              ORDER BY timestamp
            `;

            const result = await connection.runAndReadAll(query);
            const rows = result.getRowObjects() as Array<{
              timestamp: string;
              lat: number;
              lon: number;
            }>;

            // Filter by spatial and build position results + timestamps
            const posData: Array<[Timestamp, unknown]> = [];
            spatialTimestamps = new Set<string>();
            const { bbox } = spatialFilter;

            for (const row of rows) {
              if (row.lat === null || row.lon === null) continue;

              let inArea = true;

              // Lat check
              if (row.lat < bbox.south || row.lat > bbox.north) inArea = false;

              // Lon check
              if (inArea) {
                if (bbox.west <= bbox.east) {
                  if (row.lon < bbox.west || row.lon > bbox.east)
                    inArea = false;
                } else {
                  if (row.lon < bbox.west && row.lon > bbox.east)
                    inArea = false;
                }
              }

              // Precise radius check
              if (
                inArea &&
                spatialFilter.type === 'radius' &&
                spatialFilter.centerLat !== undefined &&
                spatialFilter.centerLon !== undefined &&
                spatialFilter.radiusMeters !== undefined
              ) {
                const dist = calculateDistance(
                  row.lat,
                  row.lon,
                  spatialFilter.centerLat,
                  spatialFilter.centerLon
                );
                if (dist > spatialFilter.radiusMeters) inArea = false;
              }

              if (inArea) {
                spatialTimestamps.add(row.timestamp);
                posData.push([
                  row.timestamp as Timestamp,
                  { latitude: row.lat, longitude: row.lon },
                ]);
              }
            }

            allData[pathSpecKey(posPathSpec)] = posData;
            debug(
              `[Spatial] Bucketed ${rows.length} positions, ${posData.length} within filter`
            );
          } finally {
            connection.disconnectSync();
          }
        } catch (error) {
          debug(`[Spatial] Error in fast position query: ${error}`);
          allData[pathSpecKey(posPathSpec)] = [];
          spatialTimestamps = new Set<string>();
        }
      }
    }

    // Process each path and collect data with concurrency limiting
    // Limit concurrent queries to prevent resource exhaustion (configured in cache-defaults)
    const limiter = new ConcurrencyLimiter(CONCURRENCY.MAX_QUERIES);
    await limiter.map(pathSpecs, async pathSpec => {
      try {
        // Skip position if already handled by fast spatial bucket query
        if (
          hasPositionPath &&
          isPositionPath(pathSpec.path) &&
          allData[pathSpecKey(pathSpec)]
        ) {
          return;
        }

        // Build file patterns based on query source
        let localFilePath: string | null = null;
        let s3FilePath: string | null = null;
        let effectiveTier = tier || 'raw';

        // Honour per-path skipAggregation: AggregationService never wrote
        // 5s/60s/1h files for these paths, so reading any aggregated tier
        // would return empty. Fall back to raw — the existing object-path
        // and string-path overrides further down handle their own special
        // cases and run after this.
        if (
          effectiveTier !== 'raw' &&
          this.retentionRules.shouldSkipAggregation(pathSpec.path)
        ) {
          debug(
            `Path ${pathSpec.path}: skipAggregation rule matched — overriding tier=${effectiveTier} to raw`
          );
          effectiveTier = 'raw';
        }

        // Inline filters require raw data: aggregated tiers (5s/60s/1h) blend
        // every source into each bucket and don't carry the source_* columns,
        // so they cannot be filtered.
        if (effectiveTier !== 'raw' && pathSpec.filters.length > 0) {
          debug(
            `Path ${pathSpec.path}: filter present — overriding tier=${effectiveTier} to raw`
          );
          effectiveTier = 'raw';
        }

        // Always query local first
        const sanitizedContext = this.hivePathBuilder.sanitizeContext(context);
        const sanitizedSkPath = this.hivePathBuilder.sanitizePath(
          pathSpec.path
        );
        localFilePath = path.join(
          this.dataDir,
          `tier=${effectiveTier}`,
          `context=${sanitizedContext}`,
          `path=${sanitizedSkPath}`,
          '**',
          '*.parquet'
        );
        debug(`Querying local Hive tier=${effectiveTier} at: ${localFilePath}`);

        // S3 supplements local for dates before the earliest local data
        if (this.s3Config?.enabled) {
          // Find the earliest local data date
          const localEarliestDate = this.hivePathBuilder.findEarliestDate(
            this.dataDir,
            effectiveTier,
            sanitizedContext,
            sanitizedSkPath
          );

          if (localEarliestDate && fromDate < localEarliestDate) {
            // Only query S3 for the range before local data starts
            const s3ToDate = new Date(localEarliestDate.getTime() - 86400000); // day before local starts
            if (fromDate <= s3ToDate) {
              s3FilePath = this.hivePathBuilder.buildS3Glob(
                this.s3Config.bucket,
                this.s3Config.keyPrefix || '',
                effectiveTier,
                context,
                pathSpec.path,
                fromDate,
                s3ToDate
              );
              debug(
                `S3 supplement for ${fromDate.toISOString()} to ${s3ToDate.toISOString()}`
              );
            }
          } else if (!localEarliestDate) {
            // No local data at all — query S3 for full range
            s3FilePath = this.hivePathBuilder.buildS3Glob(
              this.s3Config.bucket,
              this.s3Config.keyPrefix || '',
              effectiveTier,
              context,
              pathSpec.path,
              fromDate,
              toDate
            );
            debug(`No local data, querying S3 for full range`);
          }
        }

        // Convert ZonedDateTime to ISO string format matching parquet schema
        const fromIso = from.toInstant().toString();
        const toIso = to.toInstant().toString();

        // Get connection from pool (spatial extension already loaded), then
        // stage this path's buffer rows into a temp table for federation
        const hasBuffer = DuckDBPool.isSQLiteBufferInitialized();
        const connection = await DuckDBPool.getConnection();

        try {
          const stagedBufferTable =
            hasBuffer && this.sqliteBuffer
              ? await stageBufferTable(
                  connection,
                  this.sqliteBuffer,
                  String(context),
                  pathSpec.path,
                  fromIso,
                  toIso,
                  debug
                )
              : null;
          // Build FROM clause based on available sources
          // For hybrid queries, we UNION local and S3 sources
          // Local files need filename filter to exclude processed/quarantine/etc directories
          const buildFromClause = (filePath: string): string => {
            const isS3 = filePath.startsWith('s3://');
            if (isS3) {
              return `read_parquet('${filePath}', union_by_name=true, filename=true)`;
            }
            // Local files: exclude processed, quarantine, failed, repaired directories
            return `(SELECT * FROM read_parquet('${filePath}', union_by_name=true, filename=true) WHERE filename NOT LIKE '%/processed/%' AND filename NOT LIKE '%/quarantine/%' AND filename NOT LIKE '%/failed/%' AND filename NOT LIKE '%/repaired/%')`;
          };

          // Build FROM clause: local-only by default, hybrid only if S3 has data
          let fromClause: string;
          let localFromClause = localFilePath
            ? buildFromClause(localFilePath)
            : null;

          if (s3FilePath && localFromClause) {
            // Try hybrid: UNION local + S3, fall back to local if S3 glob has no files
            fromClause = `(
              SELECT * FROM ${localFromClause}
              UNION ALL
              SELECT * FROM ${buildFromClause(s3FilePath)}
            )`;
            debug(`Hybrid query: combining local and S3 sources`);
          } else if (s3FilePath) {
            fromClause = buildFromClause(s3FilePath);
          } else if (localFromClause) {
            fromClause = localFromClause;
          } else {
            debug(`No data source available for path ${pathSpec.path}`);
            allData[pathSpecKey(pathSpec)] = [];
            return;
          }

          // Run query with S3 fallback — if hybrid/S3 query fails, retry local-only
          const runQueryWithFallback = async (
            buildQuery: (fc: string) => string
          ): Promise<any> => {
            try {
              return await connection.runAndReadAll(buildQuery(fromClause));
            } catch (err) {
              if (s3FilePath && localFromClause) {
                debug(
                  `Hybrid query failed, falling back to local-only: ${err}`
                );
                return await connection.runAndReadAll(
                  buildQuery(localFromClause)
                );
              }
              throw err;
            }
          };

          // Check if this path has object components (value_latitude, value_longitude, etc.)
          // Use local path for schema check (S3 schema should match)
          const schemaCheckPath = localFilePath || s3FilePath;
          const componentSchema = schemaCheckPath
            ? await getPathComponentSchema(this.dataDir, context, pathSpec.path)
            : null;

          if (componentSchema && componentSchema.components.size > 0) {
            // Object path with multiple components - aggregate each component separately
            // Aggregated tiers (5s/60s/1h) collapse object paths into scalar value_avg,
            // so always query raw tier for object paths
            if (effectiveTier !== 'raw') {
              debug(
                `Path ${pathSpec.path}: Object path — overriding tier=${effectiveTier} to raw`
              );
              localFilePath = path.join(
                this.dataDir,
                'tier=raw',
                `context=${sanitizedContext}`,
                `path=${sanitizedSkPath}`,
                '**',
                '*.parquet'
              );
              // Rebuild S3 path with raw tier too
              let rawS3FilePath: string | null = null;
              if (this.s3Config?.enabled) {
                const rawEarliestDate = this.hivePathBuilder.findEarliestDate(
                  this.dataDir,
                  'raw',
                  sanitizedContext,
                  sanitizedSkPath
                );
                if (rawEarliestDate && fromDate < rawEarliestDate) {
                  const s3ToDate = new Date(
                    rawEarliestDate.getTime() - 86400000
                  );
                  if (fromDate <= s3ToDate) {
                    rawS3FilePath = this.hivePathBuilder.buildS3Glob(
                      this.s3Config.bucket,
                      this.s3Config.keyPrefix || '',
                      'raw',
                      context,
                      pathSpec.path,
                      fromDate,
                      s3ToDate
                    );
                  }
                } else if (!rawEarliestDate) {
                  rawS3FilePath = this.hivePathBuilder.buildS3Glob(
                    this.s3Config.bucket,
                    this.s3Config.keyPrefix || '',
                    'raw',
                    context,
                    pathSpec.path,
                    fromDate,
                    toDate
                  );
                }
              }
              // Rebuild fromClause with raw tier local + S3
              const rawLocalFrom = buildFromClause(localFilePath);
              localFromClause = rawLocalFrom; // Update fallback for S3 failure
              if (rawS3FilePath) {
                fromClause = `(
                  SELECT * FROM ${rawLocalFrom}
                  UNION ALL
                  SELECT * FROM ${buildFromClause(rawS3FilePath)}
                )`;
                debug(
                  `Hybrid query (raw tier): combining local and S3 sources`
                );
              } else {
                fromClause = rawLocalFrom;
              }
            }
            debug(
              `Path ${pathSpec.path}: Object path with ${componentSchema.components.size} components`
            );
            objectPaths.add(pathSpec.path); // Mark as object path

            // Build SELECT clause with one aggregate per component
            const componentSelects = Array.from(
              componentSchema.components.values()
            )
              .map(comp => {
                const aggFunc = getComponentAggregateFunction(
                  pathSpec.aggregateMethod,
                  comp.dataType
                );
                // TRY_CAST handles mixed-type parquet files (some store lat/lon as VARCHAR)
                const colExpr =
                  comp.dataType === 'numeric'
                    ? `TRY_CAST(${comp.columnName} AS DOUBLE)`
                    : comp.columnName;
                return `${aggFunc}(${colExpr}) as ${comp.name}`;
              })
              .join(',\n              ');

            // Build WHERE clause to check for at least one non-null component
            const componentWhereConditions = Array.from(
              componentSchema.components.values()
            )
              .map(comp => `${comp.columnName} IS NOT NULL`)
              .join(' OR ');

            // Check if this is a position path and spatial filter applies
            const applyPositionSpatialFilter =
              spatialFilter &&
              isPositionPath(pathSpec.path) &&
              componentSchema.components.has('latitude') &&
              componentSchema.components.has('longitude');
            const spatialWhereClause = applyPositionSpatialFilter
              ? ` AND ${buildSpatialSqlClause(spatialFilter!)}`
              : '';

            if (applyPositionSpatialFilter) {
              debug(
                `Applying spatial filter to position path ${pathSpec.path}`
              );
            }

            // Each source does its own aggregation into (timestamp, components..., priority).
            // UNION ALL the results, then ROW_NUMBER to pick highest-priority source per bucket.
            // Priority: buffer(3) > raw tier gap(2) > tier parquet(1).
            const objTsCol = getTierTimestampColumn('raw');
            const componentCols = Array.from(
              componentSchema.components.values()
            )
              .map(c => c.columnName)
              .join(', ');
            const objBucketExpr = (col: string) =>
              `strftime(DATE_TRUNC('seconds', EPOCH_MS(CAST(FLOOR(EPOCH_MS(${col}::TIMESTAMP) / ${timeResolutionMillis}) * ${timeResolutionMillis} AS BIGINT))), '%Y-%m-%dT%H:%M:%SZ')`;

            // When filtering, probe the (raw) parquet schema for the filter
            // columns so we either filter on them or exclude the parquet side
            // when absent (legacy/imported files). Probe both local and S3
            // since either may hold the data being queried.
            const objAvailable = await availableFilterColumns(
              connection,
              [localFilePath, s3FilePath],
              pathSpec.filters
            );
            const objSourceFilter = buildParquetFilterClause(
              pathSpec.filters,
              objAvailable
            );

            const buildObjQuery = (fc: string) => {
              const subqueries: string[] = [];

              // Source 1: tier parquet
              subqueries.push(`
                SELECT ${objBucketExpr(objTsCol)} as timestamp, ${componentSelects}, 1 as priority
                FROM ${fc} AS source_data
                WHERE ${objTsCol} >= '${fromIso}' AND ${objTsCol} < '${toIso}' AND (${componentWhereConditions})${spatialWhereClause}${objSourceFilter}
                GROUP BY timestamp`);

              // Source 2: SQLite buffer (today's live data not yet exported)
              if (stagedBufferTable) {
                const bufferTableCols = this.sqliteBuffer?.getTableColumns(
                  pathSpec.path
                );
                const bufferSubquery = buildBufferObjectSubquery(
                  stagedBufferTable,
                  context,
                  fromIso,
                  toIso,
                  componentSchema.components,
                  bufferTableCols,
                  pathSpec.filters
                );
                subqueries.push(`
                SELECT ${objBucketExpr('signalk_timestamp')} as timestamp, ${componentSelects}, 2 as priority
                FROM ${bufferSubquery} AS source_data
                WHERE (${componentWhereConditions})${spatialWhereClause}
                GROUP BY timestamp`);
              }

              if (subqueries.length === 1) {
                return `SELECT timestamp, ${componentCols} FROM (${subqueries[0]}) ORDER BY timestamp`;
              }

              // Pick highest priority per timestamp bucket
              const compNames = Array.from(
                componentSchema.components.keys()
              ).join(', ');
              return `
              SELECT timestamp, ${compNames} FROM (
                SELECT timestamp, ${compNames},
                  ROW_NUMBER() OVER (PARTITION BY timestamp ORDER BY priority DESC) as rn
                FROM (${subqueries.join('\n              UNION ALL')})
              ) WHERE rn = 1
              ORDER BY timestamp`;
            };

            const result = await runQueryWithFallback(buildObjQuery);
            const rows = result.getRowObjects();

            // Reconstruct objects from aggregated components
            const pathData: Array<[Timestamp, unknown]> = rows.map(
              (row: any) => {
                const timestamp = row.timestamp as Timestamp;
                const reconstructedObject: any = {};

                // Build object from component values
                componentSchema.components.forEach((comp, componentName) => {
                  const value = (row as any)[componentName];
                  if (value !== null && value !== undefined) {
                    reconstructedObject[componentName] = value;
                  }
                });

                return [timestamp, reconstructedObject];
              }
            );

            allData[pathSpecKey(pathSpec)] = pathData;
          } else {
            // Scalar path
            // First, check if value_json column exists in the parquet files
            let hasValueJson = false;
            if (localFilePath) {
              try {
                const schemaQuery = `SELECT * FROM parquet_schema('${localFilePath}') WHERE name = 'value_json'`;
                const schemaResult =
                  await connection.runAndReadAll(schemaQuery);
                hasValueJson = schemaResult.getRowObjects().length > 0;
              } catch {
                hasValueJson = false;
              }
            }

            debug(
              `Path ${pathSpec.path}: value_json column ${hasValueJson ? 'exists' : 'does not exist'}`
            );

            // Filter the parquet tier when requested. Probe both local and S3
            // (data may live in either) for the filter columns; columns absent
            // everywhere become `AND 1=0` so the parquet side contributes
            // nothing and the always-tagged buffer answers the query.
            const scalarAvailable = await availableFilterColumns(
              connection,
              [localFilePath, s3FilePath],
              pathSpec.filters
            );
            const scalarSourceFilter = buildParquetFilterClause(
              pathSpec.filters,
              scalarAvailable
            );

            // Each source does its own aggregation into (timestamp, value, priority).
            // UNION ALL the results, then ROW_NUMBER to pick highest-priority source per bucket.
            // Priority: buffer(3) > raw tier gap(2) > aggregated/raw tier parquet(1).
            // String paths must always use raw tier (no aggregated value_avg exists)
            const scalarEffectiveTier = isStringPath(pathSpec.path)
              ? 'raw'
              : effectiveTier;
            if (isStringPath(pathSpec.path) && effectiveTier !== 'raw') {
              // Rebuild fromClause pointing to raw tier for string paths
              localFilePath = path.join(
                this.dataDir,
                'tier=raw',
                `context=${sanitizedContext}`,
                `path=${sanitizedSkPath}`,
                '**',
                '*.parquet'
              );
              localFromClause = buildFromClause(localFilePath);
              fromClause = localFromClause;
            }
            const tsCol = getTierTimestampColumn(scalarEffectiveTier);
            const whereClause = getTierWhereClause(
              scalarEffectiveTier,
              hasValueJson
            );
            const bucketExpr = (col: string) =>
              `strftime(DATE_TRUNC('seconds', EPOCH_MS(CAST(FLOOR(EPOCH_MS(${col}::TIMESTAMP) / ${timeResolutionMillis}) * ${timeResolutionMillis} AS BIGINT))), '%Y-%m-%dT%H:%M:%SZ')`;

            const buildScalarQuery = (fc: string) => {
              const subqueries: string[] = [];

              // Source 1: tier parquet (uses tier-native columns and aggregate)
              const tierAggExpr = getTierAggregateExpression(
                pathSpec.aggregateMethod,
                pathSpec.path,
                scalarEffectiveTier,
                hasValueJson,
                app,
                context as string
              );
              subqueries.push(`
                SELECT ${bucketExpr(tsCol)} as timestamp, ${tierAggExpr} as value, 1 as priority
                FROM ${fc} AS source_data
                WHERE ${tsCol} >= '${fromIso}' AND ${tsCol} < '${toIso}' AND ${whereClause}${scalarSourceFilter}
                GROUP BY timestamp`);

              // Source 2: SQLite buffer (today's live data not yet exported)
              if (stagedBufferTable) {
                const bufferSubquery = buildBufferScalarSubquery(
                  stagedBufferTable,
                  context,
                  pathSpec.path,
                  fromIso,
                  toIso,
                  pathSpec.filters
                );
                const bufferAggExpr = getTierAggregateExpression(
                  pathSpec.aggregateMethod,
                  pathSpec.path,
                  'raw',
                  false,
                  app,
                  context as string
                );
                subqueries.push(`
                SELECT ${bucketExpr('signalk_timestamp')} as timestamp, ${bufferAggExpr} as value, 2 as priority
                FROM ${bufferSubquery} AS source_data
                WHERE value IS NOT NULL
                GROUP BY timestamp`);
              }

              if (subqueries.length === 1) {
                // Single source — no priority logic needed
                return `SELECT timestamp, value FROM (${subqueries[0]}) ORDER BY timestamp`;
              }

              // UNION ALL pre-aggregated results, pick highest priority per timestamp bucket
              return `
              SELECT timestamp, value FROM (
                SELECT timestamp, value,
                  ROW_NUMBER() OVER (PARTITION BY timestamp ORDER BY priority DESC) as rn
                FROM (${subqueries.join('\n              UNION ALL')})
              ) WHERE rn = 1
              ORDER BY timestamp`;
            };

            const result = await runQueryWithFallback(buildScalarQuery);
            const rows = result.getRowObjects();

            const pathData: Array<[Timestamp, unknown]> = rows.map(
              (row: any) => {
                const rowData = row as {
                  timestamp: Timestamp;
                  value: unknown;
                  value_json?: string;
                };
                const { timestamp } = rowData;
                const value = rowData.value_json
                  ? JSON.parse(String(rowData.value_json))
                  : rowData.value;
                return [timestamp, value];
              }
            );

            allData[pathSpecKey(pathSpec)] = pathData;
          }
        } finally {
          connection.disconnectSync();
        }
      } catch (error) {
        debug(`Error querying path ${pathSpec.path}: ${error}`);

        // Fallback: if parquet failed but buffer is available, query buffer only
        if (DuckDBPool.isSQLiteBufferInitialized()) {
          try {
            const fallbackFromIso = from.toInstant().toString();
            const fallbackToIso = to.toInstant().toString();
            const bufferConn = await DuckDBPool.getConnection();
            try {
              const stagedFallbackTable = this.sqliteBuffer
                ? await stageBufferTable(
                    bufferConn,
                    this.sqliteBuffer,
                    String(context),
                    pathSpec.path,
                    fallbackFromIso,
                    fallbackToIso,
                    debug
                  )
                : null;
              // Object buffer tables have value_json/value_* columns and no
              // `value` column — the scalar builder would fail against them
              const fallbackSchema = this.sqliteBuffer?.getTableSchema(
                pathSpec.path
              );
              const fallbackComponents = new Map<string, ComponentInfo>();
              if (fallbackSchema?.some(col => col.name === 'value_json')) {
                for (const col of fallbackSchema) {
                  if (
                    col.name.startsWith('value_') &&
                    col.name !== 'value_json'
                  ) {
                    const name = col.name.slice('value_'.length);
                    // Per-path buffer tables currently declare every value_*
                    // column REAL, but don't couple this to that: match all
                    // numeric SQLite affinities so an integer-typed column can
                    // never silently degrade to a string aggregation.
                    const sqliteType = col.type.toUpperCase();
                    const isNumeric = [
                      'INT',
                      'REAL',
                      'FLOA',
                      'DOUB',
                      'NUM',
                      'DEC',
                    ].some(token => sqliteType.includes(token));
                    fallbackComponents.set(name, {
                      name,
                      columnName: col.name,
                      dataType: isNumeric ? 'numeric' : 'string',
                    });
                  }
                }
              }

              if (stagedFallbackTable && fallbackComponents.size > 0) {
                const bufferSubquery = buildBufferObjectSubquery(
                  stagedFallbackTable,
                  context,
                  fallbackFromIso,
                  fallbackToIso,
                  fallbackComponents,
                  undefined,
                  pathSpec.filters
                );
                const componentSelects = Array.from(fallbackComponents.values())
                  .map(
                    comp =>
                      `${getComponentAggregateFunction(pathSpec.aggregateMethod, comp.dataType)}(${comp.columnName}) as ${comp.name}`
                  )
                  .join(', ');
                const bufferQuery = `
                  SELECT
                    strftime(DATE_TRUNC('seconds',
                      EPOCH_MS(CAST(FLOOR(EPOCH_MS(signalk_timestamp::TIMESTAMP) / ${timeResolutionMillis}) * ${timeResolutionMillis} AS BIGINT))
                    ), '%Y-%m-%dT%H:%M:%SZ') as timestamp,
                    ${componentSelects}
                  FROM ${bufferSubquery} AS source_data
                  GROUP BY timestamp
                  ORDER BY timestamp
                `;
                const bufResult = await bufferConn.runAndReadAll(bufferQuery);
                const bufRows = bufResult.getRowObjects();
                objectPaths.add(pathSpec.path);
                allData[pathSpecKey(pathSpec)] = bufRows.map((row: any) => {
                  const reconstructedObject: Record<string, unknown> = {};
                  fallbackComponents.forEach((_comp, componentName) => {
                    const value = row[componentName];
                    if (value !== null && value !== undefined) {
                      reconstructedObject[componentName] = value;
                    }
                  });
                  return [row.timestamp as Timestamp, reconstructedObject];
                });
                debug(
                  `Buffer-only fallback (object): ${bufRows.length} rows for ${pathSpec.path}`
                );
                return;
              }

              const bufferSubquery =
                stagedFallbackTable && fallbackComponents.size === 0
                  ? buildBufferScalarSubquery(
                      stagedFallbackTable,
                      context,
                      pathSpec.path,
                      fallbackFromIso,
                      fallbackToIso,
                      pathSpec.filters
                    )
                  : null;
              if (bufferSubquery) {
                const fallbackAggExpr = isStringPath(pathSpec.path)
                  ? 'FIRST(value)'
                  : 'AVG(TRY_CAST(value AS DOUBLE))';
                const bufferQuery = `
                  SELECT
                    strftime(DATE_TRUNC('seconds',
                      EPOCH_MS(CAST(FLOOR(EPOCH_MS(signalk_timestamp::TIMESTAMP) / ${timeResolutionMillis}) * ${timeResolutionMillis} AS BIGINT))
                    ), '%Y-%m-%dT%H:%M:%SZ') as timestamp,
                    ${fallbackAggExpr} as value
                  FROM ${bufferSubquery} AS source_data
                  WHERE value IS NOT NULL
                  GROUP BY timestamp
                  ORDER BY timestamp
                `;
                const bufResult = await bufferConn.runAndReadAll(bufferQuery);
                const bufRows = bufResult.getRowObjects();
                allData[pathSpecKey(pathSpec)] = bufRows.map((row: any) => [
                  row.timestamp as Timestamp,
                  row.value,
                ]);
                debug(
                  `Buffer-only fallback: ${bufRows.length} rows for ${pathSpec.path}`
                );
              }
            } finally {
              bufferConn.disconnectSync();
            }
          } catch (bufErr) {
            debug(
              `Buffer-only fallback also failed for ${pathSpec.path}: ${bufErr}`
            );
            allData[pathSpecKey(pathSpec)] = [];
          }
        } else {
          allData[pathSpecKey(pathSpec)] = [];
        }
      }
    });

    // Apply spatial timestamp filtering to non-position paths
    // This filters data to only include times when vessel was within spatial filter
    if (spatialTimestamps !== null) {
      for (const pathSpec of pathSpecs) {
        const key = pathSpecKey(pathSpec);
        if (!isPositionPath(pathSpec.path) && allData[key]) {
          if (spatialTimestamps.size === 0) {
            // No position data in area — return empty for all correlated paths
            allData[key] = [];
            debug(
              `[Spatial Correlation] No matching positions — cleared ${pathSpec.path}`
            );
          } else {
            const originalCount = allData[key].length;
            allData[key] = allData[key].filter(([timestamp]) =>
              spatialTimestamps!.has(timestamp)
            );
            debug(
              `[Spatial Correlation] Filtered ${pathSpec.path}: ${originalCount} -> ${allData[key].length} records`
            );
          }
        }
      }
    }

    // Merge all path data into time-ordered rows
    const mergedData = this.mergePathData(allData, pathSpecs);

    // Check if any path has per-path smoothing defined
    const hasPerPathSmoothing = pathSpecs.some(
      ps => ps.smoothing !== undefined
    );

    // Determine final data and values based on smoothing mode
    let finalData: Array<[Timestamp, ...unknown[]]>;
    let finalValues: HistoryValueEntry[];

    if (hasPerPathSmoothing) {
      // Per-path smoothing mode: apply smoothing only to paths that have it defined
      // Use explicit syntax: path:sma:5 or path:ema:0.3
      finalData = this.addMovingAverages(mergedData, pathSpecs, true);
      finalValues = this.buildValuesWithMovingAverages(
        pathSpecs,
        objectPaths,
        true
      );
    } else {
      // No smoothing
      finalData = mergedData;
      finalValues = pathSpecs.map(({ path, aggregateMethod, filters }) => ({
        path,
        method: aggregateMethod,
        // Echo each filter back (e.g. sourceRef) so callers can tell what each
        // column was restricted to.
        ...filterEcho(filters),
      }));
    }

    return {
      context,
      range: {
        from: from.toString() as Timestamp,
        to: to.toString() as Timestamp,
      },
      values: finalValues,
      data: finalData,
    } as DataResult;
  }

  private mergePathData(
    allData: { [key: string]: Array<[Timestamp, unknown]> },
    pathSpecs: PathSpec[]
  ): Array<[Timestamp, ...unknown[]]> {
    // Create a map of all unique timestamps
    const timestampMap = new Map<string, unknown[]>();

    pathSpecs.forEach((pathSpec, index) => {
      const pathData = allData[pathSpecKey(pathSpec)] || [];
      pathData.forEach(([timestamp, value]) => {
        if (!timestampMap.has(timestamp)) {
          timestampMap.set(timestamp, new Array(pathSpecs.length).fill(null));
        }
        timestampMap.get(timestamp)![index] = value;
      });
    });

    // Convert to sorted array format
    return Array.from(timestampMap.entries())
      .sort(([a], [b]) => a.localeCompare(b))
      .map(([timestamp, values]) => [timestamp as Timestamp, ...values]);
  }

  private addMovingAverages(
    data: Array<[Timestamp, ...unknown[]]>,
    pathSpecs: PathSpec[],
    usePerPathSmoothing: boolean = false
  ): Array<[Timestamp, ...unknown[]]> {
    if (data.length === 0) return data;

    // Default values for global moving averages mode
    const defaultSmaPeriod = 10;
    const defaultEmaAlpha = 0.2;

    // For each column, track EMA and SMA state
    // For objects, we need to track per-component state
    interface ComponentState {
      ema: number | null;
      smaWindow: number[];
    }

    const columnStates: Map<number, Map<string, ComponentState>> = new Map();

    // Initialize state for each column
    pathSpecs.forEach((_, colIndex) => {
      columnStates.set(colIndex, new Map());
    });

    return data.map((row, _rowIndex) => {
      const [timestamp, ...values] = row;
      const enhancedValues: unknown[] = [];

      values.forEach((value, colIndex) => {
        const pathSpec = pathSpecs[colIndex];
        const hasSmoothing = pathSpec.smoothing !== undefined;

        // In per-path smoothing mode, only process paths with smoothing defined
        if (usePerPathSmoothing && !hasSmoothing) {
          enhancedValues.push(value);
          return;
        }

        // Get smoothing parameters
        const smoothingType = pathSpec.smoothing;
        const smaPeriod =
          smoothingType === 'sma' && pathSpec.smoothingParam
            ? pathSpec.smoothingParam
            : defaultSmaPeriod;
        const emaAlpha =
          smoothingType === 'ema' && pathSpec.smoothingParam
            ? pathSpec.smoothingParam
            : defaultEmaAlpha;

        // Check if this is an object value (like navigation.position)
        if (value && typeof value === 'object' && !Array.isArray(value)) {
          // Object with components - calculate smoothing for each numeric component
          const enhancedObject: any = { ...value };
          const colState = columnStates.get(colIndex)!;

          Object.entries(value).forEach(([componentName, componentValue]) => {
            if (typeof componentValue === 'number' && !isNaN(componentValue)) {
              // Get or create state for this component
              if (!colState.has(componentName)) {
                colState.set(componentName, { ema: null, smaWindow: [] });
              }
              const componentState = colState.get(componentName)!;

              if (usePerPathSmoothing) {
                // Per-path mode: only apply the specific smoothing requested
                if (smoothingType === 'ema') {
                  if (componentState.ema === null) {
                    componentState.ema = componentValue;
                  } else {
                    componentState.ema =
                      emaAlpha * componentValue +
                      (1 - emaAlpha) * componentState.ema;
                  }
                  enhancedObject[componentName] =
                    Math.round(componentState.ema * 1000) / 1000;
                } else if (smoothingType === 'sma') {
                  componentState.smaWindow.push(componentValue);
                  if (componentState.smaWindow.length > smaPeriod) {
                    componentState.smaWindow =
                      componentState.smaWindow.slice(-smaPeriod);
                  }
                  const sma =
                    componentState.smaWindow.reduce(
                      (sum, val) => sum + val,
                      0
                    ) / componentState.smaWindow.length;
                  enhancedObject[componentName] = Math.round(sma * 1000) / 1000;
                }
              } else {
                // Global mode: add both EMA and SMA as separate properties
                if (componentState.ema === null) {
                  componentState.ema = componentValue;
                } else {
                  componentState.ema =
                    defaultEmaAlpha * componentValue +
                    (1 - defaultEmaAlpha) * componentState.ema;
                }

                componentState.smaWindow.push(componentValue);
                if (componentState.smaWindow.length > defaultSmaPeriod) {
                  componentState.smaWindow =
                    componentState.smaWindow.slice(-defaultSmaPeriod);
                }
                const sma =
                  componentState.smaWindow.reduce((sum, val) => sum + val, 0) /
                  componentState.smaWindow.length;

                enhancedObject[`${componentName}_ema`] =
                  Math.round(componentState.ema * 1000) / 1000;
                enhancedObject[`${componentName}_sma`] =
                  Math.round(sma * 1000) / 1000;
              }
            }
          });

          enhancedValues.push(enhancedObject);
        } else if (typeof value === 'number' && !isNaN(value)) {
          // Scalar numeric value
          const colState = columnStates.get(colIndex)!;
          const scalarKey = '__scalar__';

          if (!colState.has(scalarKey)) {
            colState.set(scalarKey, { ema: null, smaWindow: [] });
          }
          const componentState = colState.get(scalarKey)!;

          if (usePerPathSmoothing) {
            // Check if using official SignalK syntax (smoothingOnly=true)
            // Official: path:sma:5 returns ONLY the smoothed value
            // Extension: path:average:sma:5 returns raw AND smoothed values
            const smoothingOnly = pathSpec.smoothingOnly === true;

            if (!smoothingOnly) {
              // Extension syntax: include raw value first
              enhancedValues.push(value);
            }

            if (smoothingType === 'ema') {
              if (componentState.ema === null) {
                componentState.ema = value;
              } else {
                componentState.ema =
                  emaAlpha * value + (1 - emaAlpha) * componentState.ema;
              }
              enhancedValues.push(Math.round(componentState.ema * 1000) / 1000);
            } else if (smoothingType === 'sma') {
              componentState.smaWindow.push(value);
              if (componentState.smaWindow.length > smaPeriod) {
                componentState.smaWindow =
                  componentState.smaWindow.slice(-smaPeriod);
              }
              const sma =
                componentState.smaWindow.reduce((sum, val) => sum + val, 0) /
                componentState.smaWindow.length;
              enhancedValues.push(Math.round(sma * 1000) / 1000);
            }
          } else {
            // Global mode: add value, EMA, and SMA as separate columns
            enhancedValues.push(value);

            if (componentState.ema === null) {
              componentState.ema = value;
            } else {
              componentState.ema =
                defaultEmaAlpha * value +
                (1 - defaultEmaAlpha) * componentState.ema;
            }

            componentState.smaWindow.push(value);
            if (componentState.smaWindow.length > defaultSmaPeriod) {
              componentState.smaWindow =
                componentState.smaWindow.slice(-defaultSmaPeriod);
            }
            const sma =
              componentState.smaWindow.reduce((sum, val) => sum + val, 0) /
              componentState.smaWindow.length;

            enhancedValues.push(Math.round(componentState.ema * 1000) / 1000); // EMA
            enhancedValues.push(Math.round(sma * 1000) / 1000); // SMA
          }
        } else {
          // Non-numeric, non-object values (null, string, etc.)
          const smoothingOnly = pathSpec.smoothingOnly === true;
          if (usePerPathSmoothing && smoothingOnly) {
            // Official syntax: only the smoothed value column
            enhancedValues.push(null);
          } else if (usePerPathSmoothing && hasSmoothing) {
            // Extension syntax: raw + smoothed columns
            enhancedValues.push(value);
            enhancedValues.push(null); // Smoothed value placeholder
          } else if (!usePerPathSmoothing) {
            enhancedValues.push(value);
            enhancedValues.push(null); // EMA
            enhancedValues.push(null); // SMA
          } else {
            enhancedValues.push(value);
          }
        }
      });

      return [timestamp, ...enhancedValues] as [Timestamp, ...unknown[]];
    });
  }

  private buildValuesWithMovingAverages(
    pathSpecs: PathSpec[],
    objectPaths: Set<string>,
    usePerPathSmoothing: boolean = false
  ): HistoryValueEntry[] {
    const result: HistoryValueEntry[] = [];

    pathSpecs.forEach(
      ({
        path,
        aggregateMethod,
        smoothing,
        smoothingParam,
        smoothingOnly,
        filters,
      }) => {
        // Every column derived from a filtered path carries the same filters,
        // so echo them on each emitted entry.
        const echo = filterEcho(filters);
        if (usePerPathSmoothing) {
          // Per-path smoothing mode
          // Check if using official SignalK syntax (smoothingOnly=true)
          // Official: path:sma:5 returns ONLY the smoothed value
          // Extension: path:average:sma:5 returns raw AND smoothed values

          if (!smoothingOnly) {
            // Extension syntax: add raw value entry first
            result.push({ path, method: aggregateMethod, ...echo });
          }

          // Add smoothed entry if smoothing is defined
          if (smoothing) {
            const smoothedEntry: HistoryValueEntry = {
              path,
              // For official syntax, use sma/ema as the method in response
              method: smoothingOnly
                ? (smoothing as AggregateMethod)
                : aggregateMethod,
              smoothing,
              window:
                smoothingParam !== undefined
                  ? smoothingParam
                  : smoothing === 'sma'
                    ? 10
                    : 0.2,
              ...echo,
            };
            result.push(smoothedEntry);
          }
        } else if (objectPaths.has(path)) {
          // Object path - EMA/SMA are embedded in the object as component properties
          // Just add the single path entry
          result.push({ path, method: aggregateMethod, ...echo });
        } else {
          // Scalar path - add separate entries for value, EMA, and SMA
          result.push({ path, method: aggregateMethod, ...echo });
          result.push({
            path: `${path}.ema` as Path,
            method: 'ema' as AggregateMethod,
            ...echo,
          });
          result.push({
            path: `${path}.sma` as Path,
            method: 'sma' as AggregateMethod,
            ...echo,
          });
        }
      }
    );

    return result;
  }
}

// Parses a path expression into a PathSpec.
//   'navigation.headingMagnetic'                        -> { aggregate: 'average' }
//   'navigation.headingMagnetic:max'                    -> { aggregate: 'max' }
//   'navigation.headingMagnetic:average:sma:5'          -> { aggregate: 'average', smoothing: 'sma' }
//   'navigation.headingMagnetic|n2k-on-ve.can0.115'     -> { aggregate: 'average', filters: [sourceRef] }
//   'navigation.headingMagnetic:max|n2k-on-ve.can0.115' -> { aggregate: 'max', filters: [sourceRef] }
//
// Inline filters (e.g. `|sourceRef`) are split off first via parsePathFilters,
// so the `:`-separated aggregate/smoothing parsing below operates on the bare
// path expression. See utils/path-filters.ts to add another filter.
function splitPathExpression(pathExpression: string): PathSpec {
  const { base: expr, filters } = parsePathFilters(pathExpression);

  const parts = expr.split(':');
  let aggregateMethod = (parts[1] || 'average') as AggregateMethod;

  // Validate the aggregation method
  const validMethods = [
    'average',
    'min',
    'max',
    'first',
    'last',
    'mid',
    'middle_index',
  ];

  let smoothing: 'sma' | 'ema' | undefined;
  let smoothingParam: number | undefined;
  let smoothingOnly = false;

  // Check for official SignalK spec syntax: path:sma:5 or path:ema:0.2
  // In official spec, sma/ema ARE aggregation methods (returns only smoothed value)
  if (parts[1] === 'sma' || parts[1] === 'ema') {
    smoothing = parts[1];
    smoothingOnly = true; // Official syntax: return only smoothed value
    aggregateMethod = 'average' as AggregateMethod; // SMA/EMA is based on average
    if (parts[2]) {
      smoothingParam = parseFloat(parts[2]);
      if (isNaN(smoothingParam)) smoothingParam = undefined;
    }
  } else {
    // Standard aggregation method validation
    if (parts[1] && !validMethods.includes(parts[1])) {
      aggregateMethod = 'average' as AggregateMethod;
    }

    // Parse extended smoothing syntax (parts[2]) and parameter (parts[3])
    // Extension syntax: path:aggregateMethod:smoothing:param
    // Example: navigation.speedOverGround:average:sma:5
    if (parts[2] === 'sma' || parts[2] === 'ema') {
      smoothing = parts[2];
      if (parts[3]) {
        smoothingParam = parseFloat(parts[3]);
        if (isNaN(smoothingParam)) smoothingParam = undefined;
      }
    }
  }

  return {
    path: parts[0] as Path,
    queryResultName: parts[0].replace(/\./g, '_'),
    aggregateMethod,
    aggregateFunction:
      (functionForAggregate[aggregateMethod] as string) || 'avg',
    smoothing,
    smoothingParam,
    smoothingOnly,
    filters,
  };
}

// Unique key for a parsed path expression. The same path may be requested more
// than once with a different filter, aggregate, or smoothing (e.g.
// `heading|srcA,heading|srcB`); keying intermediate results by path alone would
// collapse those into one column, so the key includes every distinguishing
// field. Fields never contain spaces (paths, aggregates, and filter values are
// sanitised upstream), so a space separator is unambiguous.
function pathSpecKey(ps: PathSpec): string {
  return [
    ps.path,
    ps.aggregateMethod,
    ps.smoothing ?? '',
    ps.smoothingParam ?? '',
    ...ps.filters.map(f => `${f.column}=${f.value}`),
  ].join(' ');
}

const functionForAggregate: { [key: string]: string } = {
  average: 'avg',
  min: 'min',
  max: 'max',
  first: 'first',
  last: 'last',
  mid: 'median',
  middle_index: 'nth_value',
} as const;

function getAggregateFunction(method: AggregateMethod): string {
  switch (method) {
    case 'average':
      return 'AVG';
    case 'min':
      return 'MIN';
    case 'max':
      return 'MAX';
    case 'first':
      return 'FIRST';
    case 'last':
      return 'LAST';
    case 'mid':
      return 'MEDIAN';
    case 'middle_index':
      return 'NTH_VALUE';
    default:
      return 'AVG';
  }
}

function getValueExpression(
  pathName: string,
  hasValueJson: boolean,
  forceRaw: boolean = false
): string {
  // For position data or other complex objects, use value_json if the column exists
  if (pathName === 'navigation.position' && hasValueJson) {
    return 'value_json';
  }

  // When forceRaw is set, return value as-is (for string paths)
  if (forceRaw) {
    return 'value';
  }

  // For numeric data, try to cast to DOUBLE, fallback to the original value
  return 'TRY_CAST(value AS DOUBLE)';
}

/**
 * Get the timestamp column name based on the tier.
 * Raw tier uses signalk_timestamp, aggregated tiers use bucket_time.
 */
function getTierTimestampColumn(tier: string): string {
  return tier === 'raw' ? 'signalk_timestamp' : 'bucket_time';
}

/**
 * Get the aggregate expression for a tier-aware query.
 * For raw tier, aggregates from raw value column.
 * For aggregated tiers, uses pre-computed value_avg (or re-aggregates from sin/cos for angular).
 */
function getTierAggregateExpression(
  method: AggregateMethod,
  pathName: string,
  tier: string,
  hasValueJson: boolean,
  app?: any,
  context?: string
): string {
  // Raw tier: use original aggregate expression
  if (tier === 'raw') {
    return getAggregateExpression(method, pathName, hasValueJson, app, context);
  }

  // Aggregated tier: use pre-computed columns
  const isAngular = app && context && isAngularPath(pathName, app, context);

  if (isAngular) {
    // Re-aggregate angular paths using pre-computed sin/cos averages
    // Weight by sample_count for correct weighted averaging
    return `ATAN2(
      SUM(COALESCE(value_sin_avg, SIN(value_avg)) * sample_count) / SUM(sample_count),
      SUM(COALESCE(value_cos_avg, COS(value_avg)) * sample_count) / SUM(sample_count)
    )`;
  }

  // For standard numeric aggregation methods on pre-aggregated data
  switch (method) {
    case 'min':
      return 'MIN(value_min)';
    case 'max':
      return 'MAX(value_max)';
    case 'average':
    case undefined:
      // Weighted average using sample_count
      return 'SUM(value_avg * sample_count) / SUM(sample_count)';
    default:
      // For other methods (first, last, median), fall back to value_avg
      return `${getAggregateFunction(method)}(value_avg)`;
  }
}

/**
 * Get the WHERE clause for null filtering based on tier.
 * Raw tier checks value column, aggregated tiers check value_avg.
 */
function getTierWhereClause(tier: string, hasValueJson: boolean): string {
  if (tier === 'raw') {
    return hasValueJson
      ? '(value IS NOT NULL OR value_json IS NOT NULL)'
      : 'value IS NOT NULL';
  }
  return 'value_avg IS NOT NULL';
}

// Root-level SignalK paths that are string properties, not numeric time series
const STRING_PATHS = new Set([
  'name',
  'mmsi',
  'uuid',
  'flag',
  'port',
  'callsignVhf',
]);

function isStringPath(pathName: string): boolean {
  // Root-level paths without dots are typically string properties
  return STRING_PATHS.has(pathName) || !pathName.includes('.');
}

function getAggregateExpression(
  method: AggregateMethod,
  pathName: string,
  hasValueJson: boolean,
  app?: any,
  context?: string
): string {
  // String paths: can't AVG/MIN/MAX — use FIRST, LAST, or FIRST for average/default
  if (isStringPath(pathName)) {
    const valueExpr = getValueExpression(pathName, hasValueJson, true);
    switch (method) {
      case 'last':
        return `LAST(${valueExpr})`;
      case 'first':
      case 'average':
      case undefined:
      default:
        return `FIRST(${valueExpr})`;
    }
  }

  const valueExpr = getValueExpression(pathName, hasValueJson);

  if (method === 'middle_index') {
    // For middle_index, use FIRST as a simple fallback for now
    // TODO: Implement proper middle index selection
    return `FIRST(${valueExpr})`;
  }

  // Use vector averaging for angular paths (heading, COG, wind direction, etc.)
  if (
    (method === 'average' || method === undefined) &&
    app &&
    context &&
    isAngularPath(pathName, app, context)
  ) {
    return `ATAN2(AVG(SIN(${valueExpr})), AVG(COS(${valueExpr})))`;
  }

  return `${getAggregateFunction(method)}(${valueExpr})`;
}

/**
 * Get the appropriate aggregate function for a component based on its data type
 * Numeric components use the requested method, non-numeric use middle_index
 */
function getComponentAggregateFunction(
  requestedMethod: AggregateMethod,
  dataType: ComponentInfo['dataType']
): string {
  // For numeric components, use the requested aggregation method
  if (dataType === 'numeric') {
    // Special case: middle_index requires window functions (NTH_VALUE)
    // Use FIRST as fallback, matching scalar path behavior
    if (requestedMethod === 'middle_index') {
      return 'FIRST';
    }
    return getAggregateFunction(requestedMethod);
  }

  // For non-numeric components (string, boolean, unknown), use FIRST
  // This ensures we get a representative value from the bucket
  return 'FIRST';
}