Initial Sheet Relay application

server.js

@@ -0,0 +1,510 @@
+import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypto';
+import express from 'express';
+import { loadConfig, publicConfig, saveConfig } from './lib/config-store.js';
+import { RateLimiter, createSheetsClient, listTabs, readTabs } from './lib/google-sheets.js';
+import { clearRequests, recentRequests, recordRequest } from './lib/request-log.js';
+
+const app = express();
+const config = await loadConfig();
+const limiters = new Map();
+const consumerLimiters = new Map();
+const cache = new Map();
+const endpointHealth = new Map();
+const refreshingPublications = new Map();
+const tabCache = new Map();
+const tabCacheMilliseconds = 5 * 60 * 1000;
+const refreshIntervalMilliseconds = 5_000;
+const startedAt = Date.now();
+const metrics = { requests: 0, errors: 0, staleResponses: 0, refreshes: 0, refreshErrors: 0, refreshDurationMilliseconds: 0 };
+
+app.set('trust proxy', config.settings.trustedProxyHops);
+
+app.use(express.json({ limit: '1mb' }));
+app.use('/api', (request, response, next) => {
+  const origin = request.get('origin');
+  if (!origin) return next();
+  if (!config.settings.corsOrigins.includes('*') && !config.settings.corsOrigins.includes(origin)) return response.status(403).json({ error: 'Origin is not permitted.' });
+  response.set('Access-Control-Allow-Origin', origin);
+  response.set('Vary', 'Origin');
+  response.set('Access-Control-Allow-Headers', 'Authorization, Content-Type');
+  response.set('Access-Control-Allow-Methods', 'GET, OPTIONS');
+  if (request.method === 'OPTIONS') return response.status(204).end();
+  next();
+});
+app.use((request, response, next) => {
+  request.startedAt = Date.now();
+  next();
+});
+app.use('/vendor/lucide', express.static('node_modules/lucide/dist/umd'));
+app.use(express.static('public'));
+
+function sameSecret(received, expected) {
+  if (!received || !expected) return false;
+  const receivedBuffer = Buffer.from(received);
+  const expectedBuffer = Buffer.from(expected);
+  return receivedBuffer.length === expectedBuffer.length && timingSafeEqual(receivedBuffer, expectedBuffer);
+}
+
+function requireAdmin(request, response, next) {
+  if (!sameSecret(request.get('x-admin-token'), config.adminToken)) {
+    return response.status(401).json({ error: 'Admin authentication required.' });
+  }
+  next();
+}
+
+function requiredText(value, label) {
+  if (!value?.trim()) throw new Error(`${label} is required.`);
+  return value.trim();
+}
+
+function spreadsheetIdFrom(value) {
+  const input = requiredText(value, 'Google Sheets URL');
+  const match = input.match(/\/spreadsheets\/d\/([a-zA-Z0-9_-]+)/);
+  if (match) return match[1];
+  if (/^[a-zA-Z0-9_-]{20,}$/.test(input)) return input;
+  throw new Error('Enter a valid Google Sheets URL.');
+}
+
+function credentialFor(id) {
+  const credential = config.credentials.find((item) => item.id === id);
+  if (!credential) throw new Error('The selected credential profile no longer exists.');
+  return credential;
+}
+
+function documentFor(id) {
+  const document = config.documents.find((item) => item.id === id);
+  if (!document) throw new Error('The selected document no longer exists.');
+  return document;
+}
+
+function limiterFor(credential) {
+  const identity = credential.mode === 'serviceAccount'
+    ? JSON.parse(credential.serviceAccountJson).client_email
+    : credential.apiKey;
+  const key = createHash('sha256').update(`${credential.mode}:${identity}`).digest('hex');
+  if (!limiters.has(key)) limiters.set(key, new RateLimiter());
+  return limiters.get(key);
+}
+
+function consumerIsLimited(publication) {
+  const now = Date.now();
+  const windowMilliseconds = 60_000;
+  const limit = Math.max(1, Number(publication.consumerRequestsPerMinute) || 120);
+  const current = consumerLimiters.get(publication.id) || { startedAt: now, count: 0 };
+  if (now - current.startedAt >= windowMilliseconds) {
+    current.startedAt = now;
+    current.count = 0;
+  }
+  if (current.count >= limit) return true;
+  current.count += 1;
+  consumerLimiters.set(publication.id, current);
+  return false;
+}
+
+async function refreshPublication(publication) {
+  if (refreshingPublications.has(publication.id)) return refreshingPublications.get(publication.id);
+  const refresh = (async () => {
+    const health = endpointHealth.get(publication.id) || {};
+    health.lastAttemptAt = new Date().toISOString();
+    try {
+      const refreshStartedAt = Date.now();
+      const document = documentFor(publication.documentId);
+      const credential = credentialFor(document.credentialId);
+      const payload = {
+        publication: publication.name,
+        document: document.name,
+        spreadsheetId: document.spreadsheetId,
+        updatedAt: new Date().toISOString(),
+        sheets: await readTabs(createSheetsClient(credential), {
+          spreadsheetId: document.spreadsheetId,
+          sheets: publication.sheets,
+          dataFormat: publication.dataFormat,
+          range: publication.range,
+          omitBlankRows: publication.omitBlankRows
+        }, limiterFor(credential), credential.requestsPerMinute)
+      };
+      cache.set(publication.id, { payload, expiresAt: Date.now() + publication.cacheSeconds * 1000 });
+      endpointHealth.set(publication.id, { ...health, lastSuccessAt: new Date().toISOString(), lastError: '' });
+      metrics.refreshes += 1;
+      metrics.refreshDurationMilliseconds += Date.now() - refreshStartedAt;
+      return payload;
+    } catch (error) {
+      endpointHealth.set(publication.id, { ...health, lastError: error.message });
+      metrics.refreshErrors += 1;
+      throw error;
+    } finally {
+      refreshingPublications.delete(publication.id);
+    }
+  })();
+  refreshingPublications.set(publication.id, refresh);
+  return refresh;
+}
+
+function endpointStatus(publication) {
+  const cached = cache.get(publication.id);
+  const health = endpointHealth.get(publication.id) || {};
+  return {
+    id: publication.id,
+    name: publication.name,
+    enabled: publication.enabled !== false,
+    cacheState: !cached ? 'empty' : cached.expiresAt > Date.now() ? 'fresh' : 'stale',
+    cacheExpiresAt: cached ? new Date(cached.expiresAt).toISOString() : null,
+    ...health
+  };
+}
+
+async function tabsFor(credential, spreadsheetId) {
+  const identity = credential.mode === 'serviceAccount'
+    ? JSON.parse(credential.serviceAccountJson).client_email
+    : credential.apiKey;
+  const key = createHash('sha256').update(`${credential.mode}:${identity}:${spreadsheetId}`).digest('hex');
+  const existing = tabCache.get(key);
+  if (existing?.expiresAt > Date.now()) return existing.result;
+
+  const result = listTabs(createSheetsClient(credential), spreadsheetId, limiterFor(credential), credential.requestsPerMinute);
+  tabCache.set(key, { expiresAt: Date.now() + tabCacheMilliseconds, result });
+  try {
+    return await result;
+  } catch (error) {
+    tabCache.delete(key);
+    throw error;
+  }
+}
+
+function auditEndpointRequest(request, publicationId, status, details = {}) {
+  metrics.requests += 1;
+  if (status >= 400) metrics.errors += 1;
+  if (details.stale) metrics.staleResponses += 1;
+  recordRequest({
+    at: new Date().toISOString(),
+    publicationId,
+    status,
+    durationMs: Date.now() - request.startedAt,
+    remoteAddress: request.ip,
+    ...details
+  }).catch((error) => console.error('Unable to write request log:', error));
+}
+
+app.get('/health', (request, response) => response.json({
+  status: 'ok',
+  uptimeSeconds: Math.floor((Date.now() - startedAt) / 1000),
+  publications: config.publications.length
+}));
+
+app.get('/metrics', requireAdmin, (request, response) => {
+  const averageRefreshSeconds = metrics.refreshes ? metrics.refreshDurationMilliseconds / metrics.refreshes / 1_000 : 0;
+  response.type('text/plain').send([
+    '# TYPE sheet_relay_requests_total counter',
+    `sheet_relay_requests_total ${metrics.requests}`,
+    '# TYPE sheet_relay_errors_total counter',
+    `sheet_relay_errors_total ${metrics.errors}`,
+    '# TYPE sheet_relay_stale_responses_total counter',
+    `sheet_relay_stale_responses_total ${metrics.staleResponses}`,
+    '# TYPE sheet_relay_refreshes_total counter',
+    `sheet_relay_refreshes_total ${metrics.refreshes}`,
+    '# TYPE sheet_relay_refresh_errors_total counter',
+    `sheet_relay_refresh_errors_total ${metrics.refreshErrors}`,
+    '# TYPE sheet_relay_refresh_duration_seconds gauge',
+    `sheet_relay_refresh_duration_seconds ${averageRefreshSeconds}`
+  ].join('\n') + '\n');
+});
+
+app.get('/admin/config', requireAdmin, (request, response) => response.json(publicConfig(config)));
+
+app.post('/admin/token/rotate', requireAdmin, async (request, response, next) => {
+  try {
+    config.adminToken = randomBytes(24).toString('base64url');
+    await saveConfig(config);
+    response.json({ token: config.adminToken });
+  } catch (error) { next(error); }
+});
+
+app.put('/admin/settings', requireAdmin, async (request, response, next) => {
+  try {
+    const corsOrigins = Array.isArray(request.body.corsOrigins)
+      ? request.body.corsOrigins.map((origin) => origin.trim()).filter(Boolean)
+      : [];
+    const trustedProxyHops = Math.max(0, Math.min(10, Number(request.body.trustedProxyHops) || 0));
+    if (corsOrigins.some((origin) => origin !== '*' && !/^https?:\/\/[^/]+$/i.test(origin))) throw new Error('CORS origins must be * or an http or https origin without a path.');
+    config.settings.corsOrigins = [...new Set(corsOrigins)];
+    config.settings.trustedProxyHops = trustedProxyHops;
+    app.set('trust proxy', trustedProxyHops);
+    await saveConfig(config);
+    response.json(config.settings);
+  } catch (error) { next(error); }
+});
+
+app.get('/admin/status', requireAdmin, async (request, response, next) => {
+  try {
+    response.json({
+      status: 'ok',
+      uptimeSeconds: Math.floor((Date.now() - startedAt) / 1000),
+      caches: { endpointEntries: cache.size, tabEntries: tabCache.size },
+      endpoints: config.publications.map(endpointStatus),
+      recentRequests: await recentRequests()
+    });
+  } catch (error) { next(error); }
+});
+
+app.get('/admin/requests', requireAdmin, async (request, response, next) => {
+  try {
+    const requests = await recentRequests(500);
+    const filtered = requests.filter((entry) => {
+      if (request.query.publicationId && entry.publicationId !== request.query.publicationId) return false;
+      if (request.query.status && String(entry.status) !== String(request.query.status)) return false;
+      if (request.query.result === 'cached' && !entry.cached) return false;
+      if (request.query.result === 'refresh' && entry.cached) return false;
+      if (request.query.result === 'stale' && !entry.stale) return false;
+      return true;
+    });
+    response.json(filtered);
+  } catch (error) { next(error); }
+});
+
+app.delete('/admin/requests', requireAdmin, async (request, response, next) => {
+  try {
+    await clearRequests();
+    response.status(204).end();
+  } catch (error) { next(error); }
+});
+
+app.get('/admin/config/export', requireAdmin, (request, response) => {
+  const exported = publicConfig(config);
+  response.json({
+    version: 1,
+    settings: exported.settings,
+    credentials: exported.credentials.map(({ configured, ...credential }) => credential),
+    documents: exported.documents,
+    publications: exported.publications.map(({ tokenConfigured, ...publication }) => ({ ...publication, enabled: false }))
+  });
+});
+
+app.post('/admin/config/import', requireAdmin, async (request, response, next) => {
+  try {
+    const imported = request.body;
+    if (!imported || !Array.isArray(imported.credentials) || !Array.isArray(imported.documents) || !Array.isArray(imported.publications)) throw new Error('Choose a valid exported configuration file.');
+    const credentialIds = new Map();
+    const credentials = imported.credentials.map((credential) => {
+      if (!['apiKey', 'serviceAccount'].includes(credential.mode)) throw new Error('Imported credential has an invalid authentication type.');
+      const id = randomUUID();
+      credentialIds.set(credential.id, id);
+      return { id, name: requiredText(credential.name, 'Imported credential name'), mode: credential.mode, apiKey: '', serviceAccountJson: '', requestsPerMinute: Math.max(1, Math.min(10_000, Number(credential.requestsPerMinute) || 45)) };
+    });
+    const documentIds = new Map();
+    const documents = imported.documents.map((document) => {
+      const credentialId = credentialIds.get(document.credentialId);
+      if (!credentialId || !Array.isArray(document.sheets) || !document.sheets.length) throw new Error('Imported documents must reference an included credential and have at least one sheet.');
+      const id = randomUUID();
+      documentIds.set(document.id, id);
+      return { id, name: requiredText(document.name, 'Imported document name'), spreadsheetId: requiredText(document.spreadsheetId, 'Imported spreadsheet ID'), credentialId, sheets: document.sheets };
+    });
+    const publications = imported.publications.map((publication) => {
+      const documentId = documentIds.get(publication.documentId);
+      if (!documentId || !Array.isArray(publication.sheets) || !publication.sheets.length) throw new Error('Imported endpoints must reference an included document and have at least one sheet.');
+      return { id: randomUUID(), name: requiredText(publication.name, 'Imported endpoint name'), documentId, sheets: publication.sheets, apiToken: randomBytes(24).toString('base64url'), cacheSeconds: Math.max(0, Math.min(86_400, Number(publication.cacheSeconds) || 0)), dataFormat: publication.dataFormat === 'objects' ? 'objects' : 'rows', range: String(publication.range || '').trim(), omitBlankRows: Boolean(publication.omitBlankRows), maxRows: Math.max(1, Math.min(100_000, Number(publication.maxRows) || 10_000)), consumerRequestsPerMinute: Math.max(1, Math.min(100_000, Number(publication.consumerRequestsPerMinute) || 120)), scheduledRefresh: Boolean(publication.scheduledRefresh), enabled: false };
+    });
+    config.settings = { corsOrigins: Array.isArray(imported.settings?.corsOrigins) ? imported.settings.corsOrigins : [], trustedProxyHops: Math.max(0, Math.min(10, Number(imported.settings?.trustedProxyHops) || 0)) };
+    config.credentials = credentials;
+    config.documents = documents;
+    config.publications = publications;
+    cache.clear();
+    endpointHealth.clear();
+    app.set('trust proxy', config.settings.trustedProxyHops);
+    await saveConfig(config);
+    response.json(publicConfig(config));
+  } catch (error) { next(error); }
+});
+
+app.post('/admin/credentials', requireAdmin, async (request, response, next) => {
+  try {
+    const { id, name, mode, apiKey, serviceAccountJson, requestsPerMinute } = request.body;
+    if (!['apiKey', 'serviceAccount'].includes(mode)) throw new Error('Choose an authentication type.');
+    const existing = config.credentials.find((item) => item.id === id);
+    const suppliedSecret = mode === 'apiKey' ? apiKey?.trim() : serviceAccountJson?.trim();
+    const secret = suppliedSecret || (existing?.mode === mode ? (mode === 'apiKey' ? existing.apiKey : existing.serviceAccountJson) : '');
+    requiredText(secret, mode === 'apiKey' ? 'Google API key' : 'Service account JSON');
+    if (mode === 'serviceAccount') JSON.parse(secret);
+    const credential = {
+      id: id || randomUUID(),
+      name: requiredText(name, 'Profile name'),
+      mode,
+      apiKey: mode === 'apiKey' ? secret : '',
+      serviceAccountJson: mode === 'serviceAccount' ? secret : '',
+      requestsPerMinute: Math.max(1, Math.min(10_000, Number(requestsPerMinute) || 45))
+    };
+    const index = config.credentials.findIndex((item) => item.id === credential.id);
+    if (index >= 0) config.credentials[index] = credential;
+    else config.credentials.push(credential);
+    await saveConfig(config);
+    response.status(201).json(publicConfig(config).credentials);
+  } catch (error) { next(error); }
+});
+
+app.delete('/admin/credentials/:id', requireAdmin, async (request, response, next) => {
+  try {
+    if (config.documents.some((document) => document.credentialId === request.params.id)) throw new Error('Remove or reassign documents using this profile first.');
+    config.credentials = config.credentials.filter((credential) => credential.id !== request.params.id);
+    limiters.delete(request.params.id);
+    await saveConfig(config);
+    response.status(204).end();
+  } catch (error) { next(error); }
+});
+
+app.post('/admin/documents/discover', requireAdmin, async (request, response, next) => {
+  try {
+    const credential = credentialFor(request.body.credentialId);
+    const spreadsheetId = spreadsheetIdFrom(request.body.url);
+    const result = await tabsFor(credential, spreadsheetId);
+    response.json({ ...result, spreadsheetId });
+  } catch (error) { next(error); }
+});
+
+app.post('/admin/documents', requireAdmin, async (request, response, next) => {
+  try {
+    const { id, name, url, credentialId, sheets } = request.body;
+    credentialFor(credentialId);
+    if (!Array.isArray(sheets) || !sheets.length) throw new Error('Select at least one sheet tab.');
+    const document = { id: id || randomUUID(), name: requiredText(name, 'Document name'), spreadsheetId: spreadsheetIdFrom(url), credentialId, sheets };
+    const index = config.documents.findIndex((item) => item.id === document.id);
+    if (index >= 0) config.documents[index] = document;
+    else config.documents.push(document);
+    await saveConfig(config);
+    response.status(201).json(publicConfig(config).documents);
+  } catch (error) { next(error); }
+});
+
+app.delete('/admin/documents/:id', requireAdmin, async (request, response, next) => {
+  try {
+    if (config.publications.some((publication) => publication.documentId === request.params.id)) throw new Error('Remove endpoints using this document first.');
+    config.documents = config.documents.filter((document) => document.id !== request.params.id);
+    await saveConfig(config);
+    response.status(204).end();
+  } catch (error) { next(error); }
+});
+
+app.post('/admin/publications', requireAdmin, async (request, response, next) => {
+  try {
+    const { id, name, documentId, sheets, cacheSeconds, dataFormat, range, omitBlankRows, maxRows, consumerRequestsPerMinute, scheduledRefresh } = request.body;
+    const document = documentFor(documentId);
+    const existing = config.publications.find((item) => item.id === id);
+    const selectedSheets = Array.isArray(sheets) && sheets.length ? sheets : document.sheets;
+    if (!selectedSheets.every((sheet) => document.sheets.some((available) => available.title === sheet.title))) throw new Error('Endpoint sheets must belong to the selected document.');
+    if (!['rows', 'objects'].includes(dataFormat || 'rows')) throw new Error('Choose a valid data format.');
+    const publication = { id: id || randomUUID(), name: requiredText(name, 'Endpoint name'), documentId, sheets: selectedSheets, apiToken: existing?.apiToken || randomBytes(24).toString('base64url'), cacheSeconds: Math.max(0, Math.min(86_400, Number(cacheSeconds) || 0)), dataFormat: dataFormat || 'rows', range: String(range || '').trim(), omitBlankRows: Boolean(omitBlankRows), maxRows: Math.max(1, Math.min(100_000, Number(maxRows) || 10_000)), consumerRequestsPerMinute: Math.max(1, Math.min(100_000, Number(consumerRequestsPerMinute) || 120)), scheduledRefresh: Boolean(scheduledRefresh), enabled: existing?.enabled !== false };
+    const index = config.publications.findIndex((item) => item.id === publication.id);
+    if (index >= 0) config.publications[index] = publication;
+    else config.publications.push(publication);
+    cache.delete(publication.id);
+    endpointHealth.delete(publication.id);
+    await saveConfig(config);
+    response.status(201).json(publicConfig(config).publications);
+  } catch (error) { next(error); }
+});
+
+app.delete('/admin/publications/:id', requireAdmin, async (request, response) => {
+  config.publications = config.publications.filter((publication) => publication.id !== request.params.id);
+  cache.delete(request.params.id);
+  consumerLimiters.delete(request.params.id);
+  endpointHealth.delete(request.params.id);
+  await saveConfig(config);
+  response.status(204).end();
+});
+
+app.post('/admin/publications/:id/rotate-token', requireAdmin, async (request, response, next) => {
+  try {
+    const publication = config.publications.find((item) => item.id === request.params.id);
+    if (!publication) throw new Error('The endpoint no longer exists.');
+    publication.apiToken = randomBytes(24).toString('base64url');
+    cache.delete(publication.id);
+    await saveConfig(config);
+    response.json({ token: publication.apiToken });
+  } catch (error) { next(error); }
+});
+
+app.post('/admin/publications/:id/refresh', requireAdmin, async (request, response, next) => {
+  try {
+    const publication = config.publications.find((item) => item.id === request.params.id);
+    if (!publication) throw new Error('The endpoint no longer exists.');
+    await refreshPublication(publication);
+    response.json(endpointStatus(publication));
+  } catch (error) { next(error); }
+});
+
+app.post('/admin/publications/:id/enabled', requireAdmin, async (request, response, next) => {
+  try {
+    const publication = config.publications.find((item) => item.id === request.params.id);
+    if (!publication) throw new Error('The endpoint no longer exists.');
+    publication.enabled = Boolean(request.body.enabled);
+    await saveConfig(config);
+    response.json({ enabled: publication.enabled });
+  } catch (error) { next(error); }
+});
+
+app.get('/api/v1/:publicationId', async (request, response, next) => {
+  try {
+    const publication = config.publications.find((item) => item.id === request.params.publicationId);
+    const token = request.get('authorization')?.replace(/^Bearer\s+/i, '');
+    if (!publication || !sameSecret(token, publication.apiToken)) {
+      auditEndpointRequest(request, request.params.publicationId, 401);
+      return response.status(401).json({ error: 'Invalid API credentials.' });
+    }
+    if (publication.enabled === false) {
+      auditEndpointRequest(request, publication.id, 503, { error: 'Endpoint is disabled.' });
+      return response.status(503).json({ error: 'This endpoint is temporarily disabled.' });
+    }
+    if (consumerIsLimited(publication)) {
+      auditEndpointRequest(request, publication.id, 429, { error: 'Consumer rate limit exceeded.' });
+      response.set('Retry-After', '60');
+      return response.status(429).json({ error: 'Consumer rate limit exceeded. Try again in a minute.' });
+    }
+    const cached = cache.get(publication.id);
+    if (cached && cached.expiresAt > Date.now()) {
+      auditEndpointRequest(request, publication.id, 200, { cached: true, stale: false });
+      return response.json({ ...cached.payload, cached: true, stale: false });
+    }
+    try {
+      const payload = await refreshPublication(publication);
+      auditEndpointRequest(request, publication.id, 200, { cached: false, stale: false });
+      response.json({ ...payload, cached: false, stale: false });
+    } catch (error) {
+      if (cached) {
+        auditEndpointRequest(request, publication.id, 200, { cached: true, stale: true, error: error.message });
+        return response.json({ ...cached.payload, cached: true, stale: true, staleAt: new Date(cached.expiresAt).toISOString() });
+      }
+      throw error;
+    }
+  } catch (error) {
+    auditEndpointRequest(request, request.params.publicationId, 502, { error: error.message });
+    response.status(502).json({ error: error.message || 'Unable to refresh endpoint data.' });
+  }
+});
+
+app.use((error, request, response, next) => {
+  console.error(error);
+  response.status(400).json({ error: error.message || 'Request failed.' });
+});
+
+const port = Number(process.env.PORT) || 3000;
+app.listen(port, () => {
+  console.log(`Google Sheets API publisher listening on http://localhost:${port}`);
+  console.log('Admin token: data/config.json -> adminToken');
+});
+
+app.get('/admin/publications/:id/token', requireAdmin, (request, response, next) => {
+  try {
+    const publication = config.publications.find((item) => item.id === request.params.id);
+    if (!publication) throw new Error('The endpoint no longer exists.');
+    response.json({ token: publication.apiToken });
+  } catch (error) { next(error); }
+});
+
+setInterval(() => {
+  for (const publication of config.publications) {
+    if (!publication.scheduledRefresh || publication.cacheSeconds <= 0) continue;
+    const cached = cache.get(publication.id);
+    const refreshLeadMilliseconds = Math.min(10_000, publication.cacheSeconds * 200);
+    if (!cached || cached.expiresAt - Date.now() <= refreshLeadMilliseconds) {
+      refreshPublication(publication).catch((error) => console.error(`Scheduled refresh failed for ${publication.name}:`, error.message));
+    }
+  }
+}, refreshIntervalMilliseconds).unref();