Hub stage 1: pure merge functions for stats, top lists and query logs

Sums dashboard counters, aligns chart series by label, merges top lists by name and
query logs by time (page N cut from the merged head of every server), and reports an
unreachable server as a warning.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

hub/merge.ts

@@ -0,0 +1,245 @@
+/*
+Pure merge functions of the hub: they take the `response` objects of several
+Technitium servers and return the one a single server would have sent, so the
+console can show "All servers" without knowing there are two.
+
+No imports, no I/O: it runs under `node --experimental-strip-types` and is tested
+on its own. The types are declared here and mirror `src/api/dashboard.ts` and
+`src/api/logs.ts` on purpose, so the hub never pulls the console's sources in.
+*/
+
+export interface ChartDataset {
+  label: string
+  data: number[]
+  backgroundColor?: string | string[]
+  [key: string]: unknown
+}
+
+export interface ChartData {
+  labelFormat?: string
+  labels: string[]
+  datasets: ChartDataset[]
+  [key: string]: unknown
+}
+
+export interface TopEntry {
+  name: string
+  hits: number
+  domain?: string
+  rateLimited?: boolean
+  [key: string]: unknown
+}
+
+export interface DashboardStats {
+  stats: Record<string, number>
+  mainChartData?: ChartData
+  queryResponseChartData?: ChartData
+  queryTypeChartData?: ChartData
+  protocolTypeChartData?: ChartData
+  topClients?: TopEntry[]
+  topDomains?: TopEntry[]
+  topBlockedDomains?: TopEntry[]
+  [key: string]: unknown
+}
+
+export interface QueryLogEntry {
+  rowNumber: number
+  timestamp: string
+  node?: string
+  [key: string]: unknown
+}
+
+export interface QueryLogPage {
+  pageNumber: number
+  totalPages: number
+  totalEntries: number
+  entries: QueryLogEntry[]
+}
+
+export type Settled<T> = { node: string; ok: true; data: T } | { node: string; ok: false; error: string }
+
+const TOP_KEYS = ['topClients', 'topDomains', 'topBlockedDomains'] as const
+const BREAKDOWN_KEYS = ['queryResponseChartData', 'queryTypeChartData', 'protocolTypeChartData'] as const
+
+/*
+Charts. A chart is a list of labels and datasets whose `data` lines up with the
+labels. Merging means: build the merged label list, then for each dataset (matched
+by its `label`) add each server's value into the slot of its label. A label a
+server lacks counts as 0.
+
+`ordered` is for the time series: the merged labels keep chronological order, new
+labels are slipped in before the next label both lists share. Without it (the
+breakdowns) the labels of the first server keep their order and new ones are
+appended.
+*/
+function mergeLabels(charts: ChartData[], ordered: boolean): string[] {
+  const labels: string[] = [...charts[0].labels]
+  for (const chart of charts.slice(1)) {
+    if (!ordered) {
+      for (const label of chart.labels) if (!labels.includes(label)) labels.push(label)
+      continue
+    }
+    // Labels before the first shared one go right before it; with none shared, all are appended.
+    const anchor = chart.labels.findIndex((label) => labels.includes(label))
+    if (anchor < 0) {
+      labels.push(...chart.labels)
+      continue
+    }
+    let pos = labels.indexOf(chart.labels[anchor])
+    for (const label of chart.labels) {
+      const at = labels.indexOf(label)
+      if (at >= 0) {
+        pos = at + 1
+      } else {
+        labels.splice(pos, 0, label)
+        pos += 1
+      }
+    }
+  }
+  return labels
+}
+
+function mergeChart(charts: (ChartData | undefined)[], ordered: boolean): ChartData | undefined {
+  const present = charts.filter((c): c is ChartData => c != null && Array.isArray(c.labels))
+  if (present.length === 0) return undefined
+
+  const labels = mergeLabels(present, ordered)
+  const slot = new Map(labels.map((label, i) => [label, i]))
+  const merged = new Map<string, ChartDataset>()
+
+  for (const chart of present) {
+    for (const dataset of chart.datasets ?? []) {
+      let target = merged.get(dataset.label)
+      if (target === undefined) {
+        // The other properties (colours...) come from the first server that has the dataset.
+        target = { ...dataset, data: labels.map(() => 0) }
+        merged.set(dataset.label, target)
+      }
+      const into = target
+      chart.labels.forEach((label, i) => {
+        const value = dataset.data?.[i]
+        if (typeof value === 'number') into.data[slot.get(label) as number] += value
+      })
+    }
+  }
+  return { ...present[0], labels, datasets: [...merged.values()] }
+}
+
+/*
+Top lists: hits are summed by name, ordered by hits (highest first) then by name,
+and cut to `limit`, which defaults to the longest input list. For a client, the
+first non-empty `domain` is kept and `rateLimited` is true if any server said so.
+*/
+export function mergeTop(lists: (TopEntry[] | undefined)[], limit?: number): TopEntry[] {
+  const present = lists.filter((l): l is TopEntry[] => Array.isArray(l))
+  const byName = new Map<string, TopEntry>()
+  for (const list of present) {
+    for (const entry of list) {
+      const seen = byName.get(entry.name)
+      if (seen === undefined) {
+        byName.set(entry.name, { ...entry })
+        continue
+      }
+      seen.hits += entry.hits
+      if (!seen.domain && entry.domain) seen.domain = entry.domain
+      if (entry.rateLimited) seen.rateLimited = true
+    }
+  }
+  const cut = limit ?? Math.max(0, ...present.map((l) => l.length))
+  return [...byName.values()]
+    .sort((a, b) => b.hits - a.hits || (a.name < b.name ? -1 : a.name > b.name ? 1 : 0))
+    .slice(0, cut)
+}
+
+/* The `dashboard/stats/getTop` response is `{ topClients | topDomains | topBlockedDomains: [...] }`. */
+export function mergeTopResponses(
+  responses: Record<string, TopEntry[] | undefined>[],
+  limit?: number,
+): Record<string, TopEntry[]> {
+  for (const key of TOP_KEYS) {
+    if (responses.some((r) => Array.isArray(r[key]))) return { [key]: mergeTop(responses.map((r) => r[key]), limit) }
+  }
+  return { topDomains: [] }
+}
+
+/*
+`dashboard/stats/get`. Every numeric counter in `stats` is summed. That includes
+`totalClients` (distinct clients), which over-counts a client seen by both servers:
+the hub has no way to know the overlap from the totals alone.
+*/
+export function mergeStats(responses: DashboardStats[]): DashboardStats {
+  if (responses.length === 0) throw new Error('Nothing to merge.')
+  const first = responses[0]
+
+  const stats: Record<string, number> = {}
+  for (const r of responses) {
+    for (const [key, value] of Object.entries(r.stats ?? {})) {
+      if (typeof value === 'number') stats[key] = (stats[key] ?? 0) + value
+    }
+  }
+
+  const merged: DashboardStats = { ...first, stats }
+  const main = mergeChart(responses.map((r) => r.mainChartData), true)
+  if (main) merged.mainChartData = main
+  for (const key of BREAKDOWN_KEYS) {
+    const chart = mergeChart(responses.map((r) => r[key]), false)
+    if (chart) merged[key] = chart
+  }
+  for (const key of TOP_KEYS) {
+    if (responses.some((r) => Array.isArray(r[key]))) merged[key] = mergeTop(responses.map((r) => r[key]))
+  }
+  return merged
+}
+
+/*
+`logs/query`. The pages are what each server answered; the entries are sorted by
+timestamp together, each tagged with the server it came from. To build page N the
+caller asks every server for ITS first N x `entriesPerPage` entries (page 1 of that
+size): page N of the merged list can only be cut from the merged head of all of them.
+The hub keeps the slice of page N. `rowNumber` continues from the page's offset, as a
+single server numbers it.
+*/
+export function mergeQueryLogs(
+  pages: { node: string; page: QueryLogPage }[],
+  opts: { pageNumber: number; entriesPerPage: number; descending?: boolean },
+): QueryLogPage {
+  const descending = opts.descending !== false
+  const entries: QueryLogEntry[] = []
+  let totalEntries = 0
+  for (const { node, page } of pages) {
+    totalEntries += page.totalEntries
+    for (const entry of page.entries) entries.push({ ...entry, node })
+  }
+
+  // Array.prototype.sort is stable: equal timestamps keep the order of the servers.
+  const time = (e: QueryLogEntry) => {
+    const t = Date.parse(e.timestamp)
+    return Number.isNaN(t) ? 0 : t
+  }
+  entries.sort((a, b) => {
+    const diff = time(a) - time(b)
+    return descending ? -diff : diff
+  })
+
+  const offset = (opts.pageNumber - 1) * opts.entriesPerPage
+  return {
+    pageNumber: opts.pageNumber,
+    totalPages: Math.ceil(totalEntries / opts.entriesPerPage),
+    totalEntries,
+    entries: entries
+      .slice(offset, offset + opts.entriesPerPage)
+      .map((entry, i) => ({ ...entry, rowNumber: offset + i + 1 })),
+  }
+}
+
+/* Settled results of the parallel calls: the data that arrived plus a warning per unreachable server. */
+export function mergeSettled<T>(results: Settled<T>[]): { data: T[]; warnings: string[] } {
+  const data: T[] = []
+  const warnings: string[] = []
+  for (const result of results) {
+    if (result.ok) data.push(result.data)
+    else warnings.push(`${result.node}: ${result.error}`)
+  }
+  if (data.length === 0) throw new Error(`No server answered. ${warnings.join('; ')}`.trim())
+  return { data, warnings }
+}