diff --git a/app/server.js b/app/server.js index b67857e..a29e87b 100644 --- a/app/server.js +++ b/app/server.js @@ -101,35 +101,112 @@ function stationDateKey(ts) { return new Date(ts).toISOString().slice(0, 10); } -let lastRecordedDateutc = 0; +// Station "day" = local calendar day, since the station resets dailyrainin at local midnight. +// Uses the timezone stamped on each reading (falls back to the last one seen / America/New_York). +let stationTzFallback = 'America/New_York'; +const dayFmtCache = new Map(); +function stationLocalDay(r) { + const tz = r.tz || stationTzFallback; + let f = dayFmtCache.get(tz); + if (!f) { + try { f = new Intl.DateTimeFormat('en-CA', { timeZone: tz }); } + catch { f = new Intl.DateTimeFormat('en-CA', { timeZone: stationTzFallback }); } + dayFmtCache.set(tz, f); + } + return f.format(new Date(r.dateutc)); +} + +// Per-day set of dateutc values already on disk. Shared by the live recorder and +// the backfill so neither can write a reading the other (or a prior run) saved. +const stationSeen = new Map(); + +function stationDayFile(day) { + return path.join(STATION_DIR, `${day}.jsonl`); +} + +function getStationSeen(day) { + let set = stationSeen.get(day); + if (set) return set; + set = new Set(); + try { + for (const l of fs.readFileSync(stationDayFile(day), 'utf8').split('\n')) { + if (!l) continue; + try { set.add(JSON.parse(l).dateutc); } catch {} + } + } catch {} + stationSeen.set(day, set); + return set; +} + +// A reading is only saved if it has a sane timestamp and isn't already stored. +// Returns true when a new row was written. +function saveStationReading(obs) { + const ts = obs && obs.dateutc; + if (!Number.isFinite(ts) || ts <= 0 || ts > Date.now() + 5 * 60 * 1000) return false; + const day = stationDateKey(ts); + const seen = getStationSeen(day); + if (seen.has(ts)) return false; + fs.appendFileSync(stationDayFile(day), JSON.stringify(obs) + '\n'); + seen.add(ts); + return true; +} function recordStationReading(obs) { - if (!obs.dateutc || obs.dateutc === lastRecordedDateutc) return; try { - const day = stationDateKey(obs.dateutc); - const file = path.join(STATION_DIR, `${day}.jsonl`); - fs.appendFileSync(file, JSON.stringify(obs) + '\n'); - lastRecordedDateutc = obs.dateutc; + if (obs && obs.tz) stationTzFallback = obs.tz; + saveStationReading(obs); } catch (e) { console.error('[station] record error:', e.message); } } +// One-time startup pass: drop duplicate dateutc rows from existing day files. +// Rewrites a file (tmp + rename) only when it actually contains duplicates. +function dedupeStationFiles() { + let files; + try { files = fs.readdirSync(STATION_DIR).filter(f => f.endsWith('.jsonl')); } catch { return; } + for (const f of files) { + try { + const file = path.join(STATION_DIR, f); + const seen = new Set(); + const keep = []; + let dropped = 0; + for (const l of fs.readFileSync(file, 'utf8').split('\n')) { + if (!l) continue; + let ts; + try { ts = JSON.parse(l).dateutc; } catch { keep.push(l); continue; } + if (ts != null && seen.has(ts)) { dropped++; continue; } + seen.add(ts); + keep.push(l); + } + if (dropped) { + fs.writeFileSync(file + '.tmp', keep.join('\n') + '\n'); + fs.renameSync(file + '.tmp', file); + console.log(`[station] ${f}: removed ${dropped} duplicate reading(s)`); + } + } catch (e) { + console.error('[station] dedupe error:', f, e.message); + } + } +} +dedupeStationFiles(); + function loadStationReadings(rangeMs) { const now = Date.now(); const cutoff = now - rangeMs; const days = Math.ceil(rangeMs / 86400000) + 1; const results = []; + const seen = new Set(); for (let i = 0; i < days; i++) { const day = stationDateKey(now - i * 86400000); - const file = path.join(STATION_DIR, `${day}.jsonl`); + const file = stationDayFile(day); try { const lines = fs.readFileSync(file, 'utf8').split('\n').filter(Boolean); for (const line of lines) { try { const obj = JSON.parse(line); const ts = obj.dateutc || 0; - if (ts >= cutoff) results.push(obj); + if (ts >= cutoff && !seen.has(ts)) { seen.add(ts); results.push(obj); } } catch {} } } catch {} @@ -1114,19 +1191,6 @@ async function runBackfill(daysBack) { const cutoff = Date.now() - daysBack * 86400000; backfillState = { running: true, done: 0, total: daysBack, inserted: 0, error: null }; - // Per-day cache of existing dateutc values so we don't write duplicates - const existingByDay = new Map(); - function getExisting(day) { - if (existingByDay.has(day)) return existingByDay.get(day); - const set = new Set(); - try { - const lines = fs.readFileSync(path.join(STATION_DIR, `${day}.jsonl`), 'utf8').split('\n').filter(Boolean); - for (const l of lines) { try { set.add(JSON.parse(l).dateutc); } catch {} } - } catch {} - existingByDay.set(day, set); - return set; - } - let endDate = Date.now(); let inserted = 0; @@ -1144,13 +1208,7 @@ async function runBackfill(daysBack) { if (!rec.dateutc) continue; if (rec.dateutc < oldest) oldest = rec.dateutc; if (rec.dateutc < cutoff) continue; - const day = stationDateKey(rec.dateutc); - const existing = getExisting(day); - if (!existing.has(rec.dateutc)) { - fs.appendFileSync(path.join(STATION_DIR, `${day}.jsonl`), JSON.stringify(rec) + '\n'); - existing.add(rec.dateutc); - inserted++; - } + if (saveStationReading(rec)) inserted++; } backfillState.done = Math.min(daysBack, Math.ceil((Date.now() - oldest) / 86400000)); @@ -1252,7 +1310,7 @@ app.get('/api/station/rain-totals', requireDashboardAuth, (_req, res) => { const byDay = {}; for (const r of readings) { if (r.dailyrainin == null || !r.dateutc) continue; - const d = stationDateKey(r.dateutc); + const d = stationLocalDay(r); if (byDay[d] == null || r.dailyrainin > byDay[d]) byDay[d] = r.dailyrainin; } return +Object.values(byDay).reduce((a, b) => a + b, 0).toFixed(2); @@ -1272,7 +1330,7 @@ app.get('/api/station/degree-days', requireDashboardAuth, (req, res) => { const byDay = {}; for (const r of readings) { if (r.tempf == null || !r.dateutc) continue; - const d = stationDateKey(r.dateutc); + const d = stationLocalDay(r); if (!byDay[d]) byDay[d] = []; byDay[d].push(r.tempf); } @@ -1387,7 +1445,8 @@ app.get('/api/station/monthly-summary', requireDashboardAuth, (_req, res) => { try { const r = JSON.parse(line); if (!r.dateutc) continue; - const key = new Date(r.dateutc).toISOString().slice(0, 7); // YYYY-MM + const day = stationLocalDay(r); + const key = day.slice(0, 7); // YYYY-MM (local) if (!months[key]) months[key] = { temps: [], highTemp: null, highGust: null, dailyRainMax: {} }; const m = months[key]; if (r.tempf != null) { @@ -1396,7 +1455,6 @@ app.get('/api/station/monthly-summary', requireDashboardAuth, (_req, res) => { } if (r.windgustmph != null && (m.highGust == null || r.windgustmph > m.highGust)) m.highGust = r.windgustmph; if (r.dailyrainin != null) { - const day = stationDateKey(r.dateutc); if (m.dailyRainMax[day] == null || r.dailyrainin > m.dailyRainMax[day]) m.dailyRainMax[day] = r.dailyrainin; } } catch {} @@ -1473,7 +1531,7 @@ app.get('/api/station/streak', requireDashboardAuth, (_req, res) => { const byDay = {}; for (const r of readings) { if (r.dailyrainin == null || !r.dateutc) continue; - const d = stationDateKey(r.dateutc); + const d = stationLocalDay(r); if (byDay[d] == null || r.dailyrainin > byDay[d]) byDay[d] = r.dailyrainin; } const days = Object.keys(byDay).sort().reverse();