feat: persist push subscriptions in sqlite

lib/push-store.ts

@@ -1,83 +1,265 @@
+import Database from "better-sqlite3";
+import fs from "node:fs";
+import path from "node:path";
+import { DEFAULT_TEMPERATURE_UNIT, DEFAULT_WIND_SPEED_UNIT, isTemperatureUnit, isWindSpeedUnit } from "@/lib/weather-utils";
 import type { WarningPushSubscription } from "@/types/notifications";
+import type { Province } from "@/types/province";
 
-interface PushStore {
-  subscriptions: Map<string, WarningPushSubscription>;
-  sentWarningIds: Map<string, Set<string>>;
-  sentMorningBriefDates: Map<string, Set<string>>;
-  sentTomorrowBriefDates: Map<string, Set<string>>;
+interface PushDatabase {
+  database: Database.Database;
 }
 
-const globalStore = globalThis as typeof globalThis & { __wtrPushStore?: PushStore };
+interface PushSubscriptionRow {
+  endpoint: string;
+  subscription_json: string;
+  province: string;
+  language: string;
+  enabled: number;
+  morning_brief_enabled: number;
+  tomorrow_brief_enabled: number;
+  latitude: number | null;
+  longitude: number | null;
+  location_name: string | null;
+  county_teryt: string | null;
+  temperature_unit: string | null;
+  wind_speed_unit: string | null;
+  created_at: string;
+  updated_at: string;
+}
 
-function getStore() {
-  if (!globalStore.__wtrPushStore) {
-    globalStore.__wtrPushStore = {
-      subscriptions: new Map(),
-      sentWarningIds: new Map(),
-      sentMorningBriefDates: new Map(),
-      sentTomorrowBriefDates: new Map(),
-    };
+const globalStore = globalThis as typeof globalThis & { __wtrPushDatabase?: PushDatabase };
+
+function getDatabasePath() {
+  const configuredPath = process.env.WTR_DATABASE_PATH;
+  if (configuredPath) {
+    return path.isAbsolute(configuredPath)
+      ? configuredPath
+      : path.join(/* turbopackIgnore: true */ process.cwd(), configuredPath);
+  }
+  return path.join(/* turbopackIgnore: true */ process.cwd(), "data", "wtr.sqlite");
+}
+
+function ensureDatabaseDirectory(databasePath: string) {
+  fs.mkdirSync(path.dirname(databasePath), { recursive: true });
+}
+
+function initializeDatabase(database: Database.Database) {
+  database.pragma("journal_mode = WAL");
+  database.pragma("foreign_keys = ON");
+  database.exec(`
+    CREATE TABLE IF NOT EXISTS push_subscriptions (
+      endpoint TEXT PRIMARY KEY,
+      subscription_json TEXT NOT NULL,
+      province TEXT NOT NULL,
+      language TEXT NOT NULL,
+      enabled INTEGER NOT NULL DEFAULT 1,
+      morning_brief_enabled INTEGER NOT NULL DEFAULT 0,
+      tomorrow_brief_enabled INTEGER NOT NULL DEFAULT 0,
+      latitude REAL,
+      longitude REAL,
+      location_name TEXT,
+      county_teryt TEXT,
+      temperature_unit TEXT NOT NULL DEFAULT '${DEFAULT_TEMPERATURE_UNIT}',
+      wind_speed_unit TEXT NOT NULL DEFAULT '${DEFAULT_WIND_SPEED_UNIT}',
+      created_at TEXT NOT NULL,
+      updated_at TEXT NOT NULL
+    );
+
+    CREATE TABLE IF NOT EXISTS sent_warnings (
+      endpoint TEXT NOT NULL,
+      warning_id TEXT NOT NULL,
+      sent_at TEXT NOT NULL,
+      PRIMARY KEY (endpoint, warning_id),
+      FOREIGN KEY (endpoint) REFERENCES push_subscriptions(endpoint) ON DELETE CASCADE
+    );
+
+    CREATE TABLE IF NOT EXISTS sent_morning_briefs (
+      endpoint TEXT NOT NULL,
+      date_key TEXT NOT NULL,
+      sent_at TEXT NOT NULL,
+      PRIMARY KEY (endpoint, date_key),
+      FOREIGN KEY (endpoint) REFERENCES push_subscriptions(endpoint) ON DELETE CASCADE
+    );
+
+    CREATE TABLE IF NOT EXISTS sent_tomorrow_briefs (
+      endpoint TEXT NOT NULL,
+      date_key TEXT NOT NULL,
+      sent_at TEXT NOT NULL,
+      PRIMARY KEY (endpoint, date_key),
+      FOREIGN KEY (endpoint) REFERENCES push_subscriptions(endpoint) ON DELETE CASCADE
+    );
+  `);
+}
+
+function getDatabase() {
+  if (!globalStore.__wtrPushDatabase) {
+    const databasePath = getDatabasePath();
+    ensureDatabaseDirectory(databasePath);
+    const database = new Database(databasePath);
+    initializeDatabase(database);
+    globalStore.__wtrPushDatabase = { database };
+  }
+  return globalStore.__wtrPushDatabase.database;
+}
+
+function parseSubscription(row: PushSubscriptionRow): WarningPushSubscription | null {
+  try {
+    const subscription = JSON.parse(row.subscription_json) as WarningPushSubscription["subscription"];
+    return {
+      endpoint: row.endpoint,
+      subscription,
+      province: row.province as Province,
+      language: row.language === "en" ? "en" : "pl",
+      enabled: row.enabled === 1,
+      morningBriefEnabled: row.morning_brief_enabled === 1,
+      tomorrowBriefEnabled: row.tomorrow_brief_enabled === 1,
+      latitude: row.latitude,
+      longitude: row.longitude,
+      locationName: row.location_name,
+      countyTeryt: row.county_teryt,
+      temperatureUnit: isTemperatureUnit(row.temperature_unit) ? row.temperature_unit : DEFAULT_TEMPERATURE_UNIT,
+      windSpeedUnit: isWindSpeedUnit(row.wind_speed_unit) ? row.wind_speed_unit : DEFAULT_WIND_SPEED_UNIT,
+      createdAt: row.created_at,
+      updatedAt: row.updated_at,
+    };
+  } catch {
+    return null;
   }
-  return globalStore.__wtrPushStore;
 }
 
 export function upsertPushSubscription(subscription: WarningPushSubscription) {
-  getStore().subscriptions.set(subscription.endpoint, subscription);
-}
-
-export function removePushSubscription(endpoint: string) {
-  const store = getStore();
-  store.subscriptions.delete(endpoint);
-  store.sentWarningIds.delete(endpoint);
-  store.sentMorningBriefDates.delete(endpoint);
-  store.sentTomorrowBriefDates.delete(endpoint);
-}
-
-export function getPushSubscriptions() {
-  return Array.from(getStore().subscriptions.values());
-}
-
-export function getPushSubscription(endpoint: string) {
-  return getStore().subscriptions.get(endpoint) ?? null;
-}
-
-export function hasSentWarning(endpoint: string, warningId: string) {
-  return getStore().sentWarningIds.get(endpoint)?.has(warningId) ?? false;
-}
-
-export function markWarningSent(endpoint: string, warningId: string) {
-  const store = getStore();
-  const sentIds = store.sentWarningIds.get(endpoint) ?? new Set<string>();
-  sentIds.add(warningId);
-  store.sentWarningIds.set(endpoint, sentIds);
-}
-
-export function pruneSentWarnings(activeWarningIds: Set<string>) {
-  getStore().sentWarningIds.forEach((sentIds) => {
-    sentIds.forEach((warningId) => {
-      if (!activeWarningIds.has(warningId)) sentIds.delete(warningId);
-    });
+  getDatabase().prepare(`
+    INSERT INTO push_subscriptions (
+      endpoint,
+      subscription_json,
+      province,
+      language,
+      enabled,
+      morning_brief_enabled,
+      tomorrow_brief_enabled,
+      latitude,
+      longitude,
+      location_name,
+      county_teryt,
+      temperature_unit,
+      wind_speed_unit,
+      created_at,
+      updated_at
+    ) VALUES (
+      @endpoint,
+      @subscriptionJson,
+      @province,
+      @language,
+      @enabled,
+      @morningBriefEnabled,
+      @tomorrowBriefEnabled,
+      @latitude,
+      @longitude,
+      @locationName,
+      @countyTeryt,
+      @temperatureUnit,
+      @windSpeedUnit,
+      @createdAt,
+      @updatedAt
+    )
+    ON CONFLICT(endpoint) DO UPDATE SET
+      subscription_json = excluded.subscription_json,
+      province = excluded.province,
+      language = excluded.language,
+      enabled = excluded.enabled,
+      morning_brief_enabled = excluded.morning_brief_enabled,
+      tomorrow_brief_enabled = excluded.tomorrow_brief_enabled,
+      latitude = excluded.latitude,
+      longitude = excluded.longitude,
+      location_name = excluded.location_name,
+      county_teryt = excluded.county_teryt,
+      temperature_unit = excluded.temperature_unit,
+      wind_speed_unit = excluded.wind_speed_unit,
+      updated_at = excluded.updated_at
+  `).run({
+    endpoint: subscription.endpoint,
+    subscriptionJson: JSON.stringify(subscription.subscription),
+    province: subscription.province,
+    language: subscription.language,
+    enabled: subscription.enabled ? 1 : 0,
+    morningBriefEnabled: subscription.morningBriefEnabled ? 1 : 0,
+    tomorrowBriefEnabled: subscription.tomorrowBriefEnabled ? 1 : 0,
+    latitude: subscription.latitude,
+    longitude: subscription.longitude,
+    locationName: subscription.locationName,
+    countyTeryt: subscription.countyTeryt,
+    temperatureUnit: subscription.temperatureUnit,
+    windSpeedUnit: subscription.windSpeedUnit,
+    createdAt: subscription.createdAt,
+    updatedAt: subscription.updatedAt,
   });
 }
 
+export function removePushSubscription(endpoint: string) {
+  getDatabase().prepare("DELETE FROM push_subscriptions WHERE endpoint = ?").run(endpoint);
+}
+
+export function getPushSubscriptions() {
+  return getDatabase()
+    .prepare("SELECT * FROM push_subscriptions ORDER BY created_at ASC")
+    .all()
+    .map((row) => parseSubscription(row as PushSubscriptionRow))
+    .filter((subscription): subscription is WarningPushSubscription => subscription !== null);
+}
+
+export function getPushSubscription(endpoint: string) {
+  const row = getDatabase().prepare("SELECT * FROM push_subscriptions WHERE endpoint = ?").get(endpoint);
+  return row ? parseSubscription(row as PushSubscriptionRow) : null;
+}
+
+export function hasSentWarning(endpoint: string, warningId: string) {
+  const row = getDatabase()
+    .prepare("SELECT 1 FROM sent_warnings WHERE endpoint = ? AND warning_id = ?")
+    .get(endpoint, warningId);
+  return Boolean(row);
+}
+
+export function markWarningSent(endpoint: string, warningId: string) {
+  getDatabase()
+    .prepare("INSERT OR IGNORE INTO sent_warnings (endpoint, warning_id, sent_at) VALUES (?, ?, ?)")
+    .run(endpoint, warningId, new Date().toISOString());
+}
+
+export function pruneSentWarnings(activeWarningIds: Set<string>) {
+  const database = getDatabase();
+  if (!activeWarningIds.size) {
+    database.prepare("DELETE FROM sent_warnings").run();
+    return;
+  }
+
+  database.transaction((warningIds: string[]) => {
+    const placeholders = warningIds.map(() => "?").join(", ");
+    database.prepare(`DELETE FROM sent_warnings WHERE warning_id NOT IN (${placeholders})`).run(...warningIds);
+  })(Array.from(activeWarningIds));
+}
+
 export function hasSentMorningBrief(endpoint: string, dateKey: string) {
-  return getStore().sentMorningBriefDates.get(endpoint)?.has(dateKey) ?? false;
+  const row = getDatabase()
+    .prepare("SELECT 1 FROM sent_morning_briefs WHERE endpoint = ? AND date_key = ?")
+    .get(endpoint, dateKey);
+  return Boolean(row);
 }
 
 export function markMorningBriefSent(endpoint: string, dateKey: string) {
-  const store = getStore();
-  const sentDates = store.sentMorningBriefDates.get(endpoint) ?? new Set<string>();
-  sentDates.add(dateKey);
-  store.sentMorningBriefDates.set(endpoint, sentDates);
+  getDatabase()
+    .prepare("INSERT OR IGNORE INTO sent_morning_briefs (endpoint, date_key, sent_at) VALUES (?, ?, ?)")
+    .run(endpoint, dateKey, new Date().toISOString());
 }
 
 export function hasSentTomorrowBrief(endpoint: string, dateKey: string) {
-  return getStore().sentTomorrowBriefDates.get(endpoint)?.has(dateKey) ?? false;
+  const row = getDatabase()
+    .prepare("SELECT 1 FROM sent_tomorrow_briefs WHERE endpoint = ? AND date_key = ?")
+    .get(endpoint, dateKey);
+  return Boolean(row);
 }
 
 export function markTomorrowBriefSent(endpoint: string, dateKey: string) {
-  const store = getStore();
-  const sentDates = store.sentTomorrowBriefDates.get(endpoint) ?? new Set<string>();
-  sentDates.add(dateKey);
-  store.sentTomorrowBriefDates.set(endpoint, sentDates);
+  getDatabase()
+    .prepare("INSERT OR IGNORE INTO sent_tomorrow_briefs (endpoint, date_key, sent_at) VALUES (?, ?, ?)")
+    .run(endpoint, dateKey, new Date().toISOString());
 }