feat: add gmail-hook monorepo with watcher, api, and web

Ship the receipt pipeline (Livin/Grab/BRI), PocketBase spendings, PM2 ecosystem, and a proper .gitignore that excludes secrets, SQLite data, and media dumps.

Co-authored-by: Cursor <cursoragent@cursor.com>

apps/watcher/src/config.js

@@ -0,0 +1,73 @@
+import { config } from "dotenv";
+import { dirname, resolve } from "node:path";
+import { fileURLToPath } from "node:url";
+
+const __dirname = dirname(fileURLToPath(import.meta.url));
+export const rootDir = resolve(__dirname, "../../..");
+export const watcherDir = resolve(__dirname, "..");
+export const defaultSqlitePath = resolve(__dirname, "../../api/data/app.db");
+
+config({ path: resolve(rootDir, ".env"), quiet: true });
+
+export const PORT = Number(process.env.WATCHER_PORT || 3001);
+export const NODE_ENV = process.env.NODE_ENV || "development";
+export const GOOGLE_CLIENT_ID = process.env.GOOGLE_CLIENT_ID || "";
+export const GOOGLE_CLIENT_SECRET = process.env.GOOGLE_CLIENT_SECRET || "";
+export const GOOGLE_REDIRECT_URI =
+  process.env.GOOGLE_REDIRECT_URI || "http://localhost:5173/callback";
+export const GOOGLE_CLOUD_PROJECT =
+  process.env.GOOGLE_CLOUD_PROJECT || process.env.GCLOUD_PROJECT || "";
+export const PUBSUB_VERIFICATION_TOKEN =
+  process.env.PUBSUB_VERIFICATION_TOKEN || "";
+export const SQLITE_PATH = process.env.SQLITE_PATH || defaultSqlitePath;
+
+export const POCKETBASE_URL = (process.env.POCKETBASE_URL || "").replace(
+  /\/$/,
+  "",
+);
+export const POCKETBASE_USER = process.env.POCKETBASE_USER || "";
+export const POCKETBASE_PASSWORD = process.env.POCKETBASE_PASSWORD || "";
+
+/** ntfy topic URL, e.g. https://n.0dev.web.id/system */
+export const NTFY_URL = (process.env.NTFY_URL || "https://n.0dev.web.id/system").replace(
+  /\/$/,
+  "",
+);
+
+export const WATCH_RENEW_MS = 12 * 60 * 60 * 1000;
+
+const PUBSUB_TOPIC_RE = /^projects\/[^/]+\/topics\/[^/]+$/;
+
+export function resolvePubsubTopic(
+  topic,
+  { projectId = GOOGLE_CLOUD_PROJECT } = {},
+) {
+  const trimmed = (topic || "").trim();
+  if (!trimmed) return "";
+  if (PUBSUB_TOPIC_RE.test(trimmed)) return trimmed;
+  if (trimmed.includes("/")) return trimmed;
+  if (!projectId) return trimmed;
+  return `projects/${projectId}/topics/${trimmed}`;
+}
+
+export const GOOGLE_PUBSUB_TOPIC = resolvePubsubTopic(
+  process.env.GOOGLE_PUBSUB_TOPIC || "",
+);
+
+export function assertConfig() {
+  const missing = [];
+  if (!GOOGLE_CLIENT_ID) missing.push("GOOGLE_CLIENT_ID");
+  if (!GOOGLE_CLIENT_SECRET) missing.push("GOOGLE_CLIENT_SECRET");
+  if (!GOOGLE_PUBSUB_TOPIC) missing.push("GOOGLE_PUBSUB_TOPIC");
+  if (NODE_ENV === "production" && !PUBSUB_VERIFICATION_TOKEN) {
+    missing.push("PUBSUB_VERIFICATION_TOKEN");
+  }
+  if (missing.length) {
+    throw new Error(`Missing required env: ${missing.join(", ")}`);
+  }
+  if (!PUBSUB_TOPIC_RE.test(GOOGLE_PUBSUB_TOPIC)) {
+    throw new Error(
+      `GOOGLE_PUBSUB_TOPIC must be a full Pub/Sub name like projects/my-gcp-project/topics/gmail-push (got "${process.env.GOOGLE_PUBSUB_TOPIC || ""}")`,
+    );
+  }
+}

apps/watcher/src/db.js

@@ -0,0 +1,102 @@
+import { mkdirSync } from "node:fs";
+import { dirname } from "node:path";
+import Database from "better-sqlite3";
+import { SQLITE_PATH } from "./config.js";
+
+mkdirSync(dirname(SQLITE_PATH), { recursive: true });
+
+const db = new Database(SQLITE_PATH);
+db.pragma("journal_mode = WAL");
+db.pragma("foreign_keys = ON");
+db.pragma("busy_timeout = 5000");
+
+db.exec(`
+  CREATE TABLE IF NOT EXISTS users (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    google_sub TEXT NOT NULL UNIQUE,
+    email TEXT NOT NULL,
+    name TEXT,
+    picture TEXT,
+    created_at TEXT NOT NULL DEFAULT (datetime('now'))
+  );
+
+  CREATE TABLE IF NOT EXISTS oauth_tokens (
+    user_id INTEGER PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
+    access_token TEXT NOT NULL,
+    refresh_token TEXT,
+    expiry INTEGER NOT NULL
+  );
+
+  CREATE TABLE IF NOT EXISTS watch_state (
+    user_id INTEGER PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
+    history_id TEXT NOT NULL,
+    expiration INTEGER NOT NULL
+  );
+`);
+
+const getUserByIdStmt = db.prepare(`SELECT * FROM users WHERE id = ?`);
+const getUserByEmailStmt = db.prepare(
+  `SELECT * FROM users WHERE LOWER(email) = LOWER(?)`,
+);
+const getTokensStmt = db.prepare(`SELECT * FROM oauth_tokens WHERE user_id = ?`);
+const listUsersWithTokensStmt = db.prepare(`
+  SELECT u.*
+  FROM users u
+  INNER JOIN oauth_tokens t ON t.user_id = u.id
+`);
+const saveTokensStmt = db.prepare(`
+  INSERT INTO oauth_tokens (user_id, access_token, refresh_token, expiry)
+  VALUES (@userId, @accessToken, @refreshToken, @expiry)
+  ON CONFLICT(user_id) DO UPDATE SET
+    access_token = excluded.access_token,
+    refresh_token = COALESCE(excluded.refresh_token, oauth_tokens.refresh_token),
+    expiry = excluded.expiry
+`);
+const getWatchStateStmt = db.prepare(
+  `SELECT * FROM watch_state WHERE user_id = ?`,
+);
+const saveWatchStateStmt = db.prepare(`
+  INSERT INTO watch_state (user_id, history_id, expiration)
+  VALUES (@userId, @historyId, @expiration)
+  ON CONFLICT(user_id) DO UPDATE SET
+    history_id = excluded.history_id,
+    expiration = excluded.expiration
+`);
+
+export function getUserById(id) {
+  return getUserByIdStmt.get(id) || null;
+}
+
+export function getUserByEmail(email) {
+  if (!email) return null;
+  return getUserByEmailStmt.get(email) || null;
+}
+
+export function getTokens(userId) {
+  return getTokensStmt.get(userId) || null;
+}
+
+export function listUsersWithTokens() {
+  return listUsersWithTokensStmt.all();
+}
+
+export function saveTokens(userId, { accessToken, refreshToken, expiry }) {
+  saveTokensStmt.run({
+    userId,
+    accessToken,
+    refreshToken: refreshToken || null,
+    expiry,
+  });
+}
+
+export function getWatchState(userId) {
+  return getWatchStateStmt.get(userId) || null;
+}
+
+export function saveWatchState(userId, { historyId, expiration }) {
+  saveWatchStateStmt.run({
+    userId,
+    historyId: String(historyId),
+    expiration: Number(expiration),
+  });
+}

apps/watcher/src/google.js

@@ -0,0 +1,105 @@
+import { google } from "googleapis";
+import {
+  GOOGLE_CLIENT_ID,
+  GOOGLE_CLIENT_SECRET,
+  GOOGLE_PUBSUB_TOPIC,
+  GOOGLE_REDIRECT_URI,
+} from "./config.js";
+import { getTokens, getWatchState, saveTokens, saveWatchState } from "./db.js";
+import { HttpError } from "./http.js";
+
+export function createOAuthClient() {
+  return new google.auth.OAuth2(
+    GOOGLE_CLIENT_ID,
+    GOOGLE_CLIENT_SECRET,
+    GOOGLE_REDIRECT_URI,
+  );
+}
+
+function persistClientTokens(userId, client) {
+  client.on("tokens", (tokens) => {
+    if (!tokens.access_token) return;
+    saveTokens(userId, {
+      accessToken: tokens.access_token,
+      refreshToken: tokens.refresh_token || null,
+      expiry: tokens.expiry_date || Date.now() + 3600 * 1000,
+    });
+  });
+}
+
+export async function gmailClientForUser(userId) {
+  const stored = getTokens(userId);
+  if (!stored?.access_token && !stored?.refresh_token) {
+    throw new HttpError(401, "Google account is not connected");
+  }
+
+  const client = createOAuthClient();
+  client.setCredentials({
+    access_token: stored.access_token,
+    refresh_token: stored.refresh_token,
+    expiry_date: stored.expiry,
+  });
+  persistClientTokens(userId, client);
+  try {
+    await client.getAccessToken();
+  } catch {
+    throw new HttpError(
+      401,
+      "Gmail access expired or was revoked. Sign in again.",
+    );
+  }
+  return google.gmail({ version: "v1", auth: client });
+}
+
+export async function startWatch(userId, { resetCursor = false } = {}) {
+  const gmail = await gmailClientForUser(userId);
+  const { data } = await gmail.users.watch({
+    userId: "me",
+    requestBody: {
+      topicName: GOOGLE_PUBSUB_TOPIC,
+      labelIds: ["INBOX"],
+    },
+  });
+
+  const existing = getWatchState(userId);
+  const historyId =
+    resetCursor || !existing?.history_id
+      ? String(data.historyId)
+      : existing.history_id;
+
+  saveWatchState(userId, {
+    historyId,
+    expiration: Number(data.expiration) || Date.now() + 7 * 24 * 3600 * 1000,
+  });
+
+  return { ...data, historyId };
+}
+
+export async function listHistory(gmail, startHistoryId) {
+  const messageIds = new Set();
+  let pageToken;
+  let latestHistoryId = String(startHistoryId);
+
+  do {
+    const { data } = await gmail.users.history.list({
+      userId: "me",
+      startHistoryId: String(startHistoryId),
+      pageToken,
+      historyTypes: ["messageAdded"],
+    });
+    if (data.historyId) latestHistoryId = String(data.historyId);
+    for (const record of data.history || []) {
+      for (const added of record.messagesAdded || []) {
+        const id = added.message?.id;
+        if (id) messageIds.add(id);
+      }
+    }
+    pageToken = data.nextPageToken;
+  } while (pageToken);
+
+  return { messageIds: [...messageIds], historyId: latestHistoryId };
+}
+
+export function googleStatus(err) {
+  return Number(err.code || err.response?.status || err.statusCode || 0);
+}

apps/watcher/src/handlers/fetch-and-log-message.js

@@ -0,0 +1,76 @@
+import {
+  extractBody,
+  extractHtml,
+  hydrateRfc822Message,
+  listRfc822Parts,
+  summarizeMessage,
+} from "../mime.js";
+import { googleStatus } from "../google.js";
+
+const BODY_PREVIEW = 500;
+
+export async function fetchAndLogMessage(ctx) {
+  const { gmail, messageId, log } = ctx;
+  try {
+    const { data } = await gmail.users.messages.get({
+      userId: "me",
+      id: messageId,
+      format: "full",
+    });
+    const message = {
+      ...summarizeMessage(data),
+      body: extractBody(data.payload),
+      html: extractHtml(data.payload),
+    };
+
+    const rfcParts = listRfc822Parts(data.payload);
+    const emlMessages = [];
+    for (let index = 0; index < rfcParts.length; index++) {
+      const part = rfcParts[index];
+      const hydrated = await hydrateRfc822Message(gmail, message.id, part, index);
+      emlMessages.push(hydrated);
+    }
+
+    const emptyEmls = emlMessages.filter((m) => !m.body && !m.html).length;
+
+    log.info(
+      {
+        messageId,
+        from: message.from,
+        subject: message.subject,
+        date: message.date,
+        snippet: message.snippet,
+        bodyPreview: (message.body || "").slice(0, BODY_PREVIEW),
+        emlCount: emlMessages.length,
+        emptyEmlCount: emptyEmls || undefined,
+      },
+      "gmail-watch: message",
+    );
+
+    if (emlMessages.length) {
+      log.info(
+        {
+          messageId,
+          emlCount: emlMessages.length,
+          filenames: emlMessages.map((m) => m.attachmentFilename),
+        },
+        "gmail-watch: eml attachments",
+      );
+    }
+
+    return {
+      pass: {
+        message,
+        emlMessages,
+        gmailPayload: data.payload,
+      },
+    };
+  } catch (err) {
+    const status = googleStatus(err);
+    if (status === 404) {
+      log.warn({ messageId }, "gmail-watch: message not found");
+      return { break: true };
+    }
+    throw err;
+  }
+}

apps/watcher/src/handlers/log-event.js

@@ -0,0 +1,12 @@
+export async function logEvent(ctx) {
+  ctx.log.info(
+    {
+      emailAddress: ctx.notification?.emailAddress,
+      historyId: ctx.notification?.historyId,
+      messageId: ctx.messageId,
+      userId: ctx.userId,
+    },
+    "gmail-watch: event",
+  );
+  return { pass: { loggedAt: Date.now() } };
+}

apps/watcher/src/handlers/match-rule.js

@@ -0,0 +1,58 @@
+import { extractEmail, resolveOriginal, subjectMatches } from "../lib/email.js";
+import { rules } from "../rules/index.js";
+
+/**
+ * Match a fetched message against known receipt rules using original From + Subject.
+ * Breaks the pipeline when nothing matches.
+ */
+export async function matchRule(ctx) {
+  const { message, log, messageId } = ctx;
+  if (!message) {
+    log.warn({ messageId }, "gmail-watch: match skipped (no message)");
+    return { break: true };
+  }
+
+  const original = resolveOriginal(message);
+  const fromEmail = extractEmail(original.from);
+
+  const rule = rules.find(
+    (r) =>
+      fromEmail === r.fromAddress.toLowerCase() &&
+      subjectMatches(original.subject, r.subject),
+  );
+
+  if (!rule) {
+    log.info(
+      {
+        messageId,
+        from: original.from,
+        subject: original.subject,
+        viaForward: original.viaForward,
+      },
+      "gmail-watch: no rule match",
+    );
+    return { break: true };
+  }
+
+  log.info(
+    {
+      messageId,
+      kind: rule.kind,
+      from: original.from,
+      subject: original.subject,
+      viaForward: original.viaForward,
+    },
+    "gmail-watch: rule matched",
+  );
+
+  return {
+    pass: {
+      rule,
+      original: {
+        from: original.from,
+        subject: original.subject,
+        date: original.date,
+      },
+    },
+  };
+}

apps/watcher/src/handlers/parse-matched-message.js

@@ -0,0 +1,76 @@
+import { parseBriTransfer } from "../parsers/bri.js";
+import { parseGrabReceipt } from "../parsers/grab.js";
+import { parseLivinReceipt } from "../parsers/livin.js";
+import { notifyReceiptIssue } from "../notify/ntfy.js";
+
+const PARSERS = {
+  "livin.qr_payment.success": parseLivinReceipt,
+  "grab.ereceipt.ride": parseGrabReceipt,
+  "bri.transfer.success": parseBriTransfer,
+};
+
+/**
+ * Parse a matched message body into the shared envelope JSON.
+ * On failure, logs a warning, notifies ntfy, and breaks (Pub/Sub stays healthy).
+ */
+export async function parseMatchedMessage(ctx) {
+  const { message, rule, original, log, messageId, notification } = ctx;
+  const parse = PARSERS[rule?.kind];
+  if (!parse) {
+    log.warn({ messageId, kind: rule?.kind }, "gmail-watch: no parser for kind");
+    await notifyReceiptIssue(
+      {
+        reason: "no parser for kind",
+        messageId,
+        kind: rule?.kind,
+        message,
+      },
+      log,
+    );
+    return { break: true };
+  }
+
+  const fields = parse(message.body || "", { html: message.html || "" });
+  if (!fields) {
+    log.warn(
+      { messageId, kind: rule.kind },
+      "gmail-watch: parse failed",
+    );
+    await notifyReceiptIssue(
+      {
+        reason: "parse failed",
+        messageId,
+        kind: rule.kind,
+        message,
+      },
+      log,
+    );
+    return { break: true };
+  }
+
+  // Grab tip shares From/Subject with ride; parser sets kind/reference.
+  const kind = fields.kind || rule.kind;
+
+  const parsed = {
+    kind,
+    messageId,
+    mailbox: notification?.emailAddress || "",
+    matched: {
+      from: message.from || "",
+      subject: message.subject || "",
+    },
+    original: {
+      from: original?.from || "",
+      subject: original?.subject || "",
+      date: original?.date || "",
+    },
+    amount: fields.amount,
+    currency: fields.currency || "IDR",
+    reference: fields.reference,
+    occurredAt: fields.occurredAt,
+    details: fields.details || {},
+  };
+
+  log.info({ messageId, kind: parsed.kind, parsed }, "gmail-watch: parsed");
+  return { pass: { parsed } };
+}

apps/watcher/src/handlers/save-spending.js

@@ -0,0 +1,126 @@
+import { pocketbaseConfigured, createRecord } from "../pocketbase/client.js";
+import { toSpendingRecord } from "../pocketbase/map.js";
+import { notifyReceiptIssue } from "../notify/ntfy.js";
+
+/**
+ * Insert a parsed receipt into PocketBase spendings.
+ * Unique invoice_id conflicts are logged and skipped (no throw).
+ * Non-duplicate skips / errors notify ntfy.
+ */
+export async function saveSpending(ctx) {
+  const { parsed, rule, log, messageId, message } = ctx;
+
+  if (!parsed) {
+    log.warn({ messageId }, "gmail-watch: save skipped (no parsed)");
+    await notifyReceiptIssue(
+      {
+        reason: "save skipped (no parsed)",
+        messageId,
+        message,
+      },
+      log,
+    );
+    return { break: true };
+  }
+
+  if (!pocketbaseConfigured()) {
+    log.warn(
+      { messageId, kind: parsed.kind },
+      "gmail-watch: pocketbase not configured; parsed only",
+    );
+    await notifyReceiptIssue(
+      {
+        reason: "pocketbase not configured",
+        messageId,
+        kind: parsed.kind,
+        message,
+        extra: { parsed },
+      },
+      log,
+    );
+    return { pass: { saved: false } };
+  }
+
+  const record = toSpendingRecord(parsed, rule);
+  if (!record.invoice_id || record.amount == null || !record.trx_date) {
+    log.warn(
+      { messageId, kind: parsed.kind, record },
+      "gmail-watch: save skipped (incomplete record)",
+    );
+    await notifyReceiptIssue(
+      {
+        reason: "save skipped (incomplete record)",
+        messageId,
+        kind: parsed.kind,
+        message,
+        record,
+        extra: { parsed },
+      },
+      log,
+    );
+    return { break: true };
+  }
+
+  try {
+    const { ok, status, data } = await createRecord("spendings", record);
+    if (ok) {
+      log.info(
+        {
+          messageId,
+          kind: parsed.kind,
+          invoice_id: record.invoice_id,
+          recordId: data.id,
+        },
+        "gmail-watch: spending saved",
+      );
+      return { pass: { saved: true, spendingId: data.id } };
+    }
+
+    // Unique constraint / duplicate — no ntfy
+    const msg = String(data?.message || "");
+    const isUnique =
+      status === 400 &&
+      (/unique|invoice_id/i.test(msg) ||
+        data?.data?.invoice_id ||
+        JSON.stringify(data).includes("unique"));
+
+    if (isUnique) {
+      log.info(
+        { messageId, invoice_id: record.invoice_id, status },
+        "gmail-watch: spending duplicate skipped",
+      );
+      return { pass: { saved: false, duplicate: true } };
+    }
+
+    log.error(
+      { messageId, status, data },
+      "gmail-watch: spending save failed",
+    );
+    await notifyReceiptIssue(
+      {
+        reason: "spending save failed",
+        messageId,
+        kind: parsed.kind,
+        message,
+        record,
+        extra: { status, data },
+      },
+      log,
+    );
+    return { pass: { saved: false, error: true } };
+  } catch (err) {
+    log.error({ err, messageId }, "gmail-watch: spending save error");
+    await notifyReceiptIssue(
+      {
+        reason: "spending save error",
+        messageId,
+        kind: parsed.kind,
+        message,
+        record,
+        extra: { error: err.message || String(err) },
+      },
+      log,
+    );
+    return { pass: { saved: false, error: true } };
+  }
+}

apps/watcher/src/http.js

@@ -0,0 +1,6 @@
+export class HttpError extends Error {
+  constructor(statusCode, message) {
+    super(message);
+    this.statusCode = statusCode;
+  }
+}

apps/watcher/src/index.js

@@ -0,0 +1,92 @@
+import Fastify from "fastify";
+import { GOOGLE_PUBSUB_TOPIC, PORT, WATCH_RENEW_MS, assertConfig } from "./config.js";
+import { listUsersWithTokens } from "./db.js";
+import { startWatch } from "./google.js";
+import { fetchAndLogMessage } from "./handlers/fetch-and-log-message.js";
+import { logEvent } from "./handlers/log-event.js";
+import { matchRule } from "./handlers/match-rule.js";
+import { parseMatchedMessage } from "./handlers/parse-matched-message.js";
+import { saveSpending } from "./handlers/save-spending.js";
+import { HttpError } from "./http.js";
+import { pocketbaseConfigured } from "./pocketbase/client.js";
+import { ensureSpendingsSchema } from "./pocketbase/schema.js";
+import { registerPubsubRoutes } from "./routes/pubsub.js";
+
+assertConfig();
+
+const fetchHandlers = [logEvent, fetchAndLogMessage];
+const processHandlers = [matchRule, parseMatchedMessage, saveSpending];
+
+const app = Fastify({
+  logger: true,
+  trustProxy: true,
+});
+
+app.get("/health", async () => ({ ok: true }));
+
+await registerPubsubRoutes(app, { fetchHandlers, processHandlers });
+
+app.setErrorHandler((err, request, reply) => {
+  const status = err.statusCode || 500;
+  if (status >= 500) {
+    request.log.error(err);
+  }
+  const message =
+    err instanceof HttpError || status < 500
+      ? err.message
+      : "Internal server error";
+  reply.code(status).send({ error: message });
+});
+
+async function renewAllWatches() {
+  const users = listUsersWithTokens();
+  if (!users.length) {
+    app.log.warn("gmail-watch: no signed-in users with tokens");
+    return;
+  }
+  for (const user of users) {
+    try {
+      await startWatch(user.id);
+      app.log.info(
+        { userId: user.id, email: user.email },
+        "gmail-watch: watch started",
+      );
+    } catch (err) {
+      app.log.error({ err, userId: user.id, email: user.email }, "gmail-watch: watch failed");
+    }
+  }
+}
+
+let renewTimer;
+app.addHook("onClose", async () => {
+  if (renewTimer) clearInterval(renewTimer);
+});
+
+try {
+  await app.listen({ port: PORT, host: "0.0.0.0" });
+  app.log.info(
+    { topic: GOOGLE_PUBSUB_TOPIC },
+    `Watcher listening on http://localhost:${PORT}`,
+  );
+
+  if (pocketbaseConfigured()) {
+    try {
+      await ensureSpendingsSchema(app.log);
+    } catch (err) {
+      app.log.error({ err }, "pocketbase: schema ensure failed");
+    }
+  } else {
+    app.log.warn("pocketbase: not configured; spendings will not be saved");
+  }
+
+  await renewAllWatches();
+  renewTimer = setInterval(() => {
+    renewAllWatches().catch((err) => {
+      app.log.error({ err }, "gmail-watch: renew cycle failed");
+    });
+  }, WATCH_RENEW_MS);
+  renewTimer.unref?.();
+} catch (err) {
+  app.log.error(err);
+  process.exit(1);
+}

apps/watcher/src/lib/email.js

@@ -0,0 +1,66 @@
+/**
+ * Extract a bare email address from a From header value.
+ * e.g. `"Livin'" <noreply.livin@bankmandiri.co.id>` → `noreply.livin@bankmandiri.co.id`
+ */
+export function extractEmail(from) {
+  if (!from) return "";
+  const angle = from.match(/<([^>]+)>/);
+  if (angle) return angle[1].trim().toLowerCase();
+  const bare = from.match(/[\w.+-]+@[\w.-]+\.\w+/);
+  return bare ? bare[0].toLowerCase() : "";
+}
+
+/**
+ * Parse a Gmail "Forwarded message" header block from the body.
+ * Returns null if no forwarded header is present.
+ */
+export function parseForwardedHeader(body) {
+  if (!body) return null;
+  const marker = /-{2,}\s*Forwarded message\s*-{2,}/i;
+  const idx = body.search(marker);
+  if (idx < 0) return null;
+
+  const after = body.slice(idx);
+  const headerMatch = after.match(
+    /From:\s*(.+)\r?\nDate:\s*(.+)\r?\nSubject:\s*(.+)\r?\nTo:\s*(.+)/i,
+  );
+  if (!headerMatch) return null;
+
+  return {
+    from: headerMatch[1].trim(),
+    date: headerMatch[2].trim(),
+    subject: headerMatch[3].trim(),
+    to: headerMatch[4].trim(),
+  };
+}
+
+/**
+ * Resolve the original sender/subject/date for matching.
+ * Prefer forwarded header when present; otherwise use the Gmail envelope.
+ */
+export function resolveOriginal(message) {
+  const forwarded = parseForwardedHeader(message?.body || "");
+  if (forwarded) {
+    return {
+      from: forwarded.from,
+      subject: forwarded.subject,
+      date: forwarded.date,
+      viaForward: true,
+    };
+  }
+  return {
+    from: message?.from || "",
+    subject: message?.subject || "",
+    date: message?.date || "",
+    viaForward: false,
+  };
+}
+
+export function subjectMatches(actual, expected) {
+  if (!actual || !expected) return false;
+  const a = actual.trim().toLowerCase();
+  const e = expected.trim().toLowerCase();
+  // Strip common Fwd:/Re: prefixes for comparison of the core subject
+  const strip = (s) => s.replace(/^(fwd|re):\s*/i, "").trim();
+  return a === e || strip(a) === strip(e) || a.includes(e);
+}

apps/watcher/src/lib/parse-helpers.js

@@ -0,0 +1,168 @@
+/** Parse Indonesian Rupiah strings like "Rp 22.000,00" or "Rp16.600" into integers. */
+export function parseIdrAmount(raw) {
+  if (raw == null) return null;
+  const s = String(raw).replace(/\s/g, "");
+  const match = s.match(/Rp?\s*([\d.]+(?:,\d{2})?)/i) || s.match(/([\d.]+(?:,\d{2})?)/);
+  if (!match) return null;
+  // Indonesian: dots are thousands separators, comma is decimal
+  const whole = match[1].replace(/\./g, "").replace(/,\d+$/, "");
+  const n = Number(whole);
+  return Number.isFinite(n) ? n : null;
+}
+
+const MONTHS = {
+  // English
+  jan: 1,
+  january: 1,
+  feb: 2,
+  february: 2,
+  mar: 3,
+  march: 3,
+  apr: 4,
+  april: 4,
+  may: 5,
+  jun: 6,
+  june: 6,
+  jul: 7,
+  july: 7,
+  aug: 8,
+  august: 8,
+  sep: 9,
+  sept: 9,
+  september: 9,
+  oct: 10,
+  october: 10,
+  nov: 11,
+  november: 11,
+  dec: 12,
+  december: 12,
+  // Indonesian (Livin uses e.g. "15 Agu 2026")
+  januari: 1,
+  februari: 2,
+  maret: 3,
+  mei: 5,
+  juni: 6,
+  juli: 7,
+  agustus: 8,
+  agu: 8,
+  ags: 8,
+  oktober: 10,
+  okt: 10,
+  nop: 11,
+  desember: 12,
+  des: 12,
+};
+
+/**
+ * Parse dates like "13 Sep 2026", "13 September 2026", or Indonesian
+ * "15 Agu 2026" into YYYY-MM-DD. Also accepts 2-digit years ("13 Sep 26").
+ */
+export function parseDayMonthYear(raw) {
+  if (!raw) return null;
+  const m = String(raw)
+    .trim()
+    .match(/(\d{1,2})\s+([A-Za-z]+)\s+(\d{2,4})\b/);
+  if (!m) return null;
+  const day = Number(m[1]);
+  const month = MONTHS[m[2].toLowerCase()];
+  let year = Number(m[3]);
+  if (m[3].length === 2) year += year >= 70 ? 1900 : 2000;
+  if (!month || !day || !year) return null;
+  return `${year}-${String(month).padStart(2, "0")}-${String(day).padStart(2, "0")}`;
+}
+
+/**
+ * Parse Grab tip timestamps like "13 Sep 26 10:47 +0700" → ISO with offset.
+ */
+export function parseGrabDateTime(raw) {
+  if (!raw) return null;
+  const m = String(raw)
+    .trim()
+    .match(
+      /(\d{1,2})\s+([A-Za-z]+)\s+(\d{2,4})\s+(\d{1,2}):(\d{2})(?::(\d{2}))?\s*([+-]\d{4})?/,
+    );
+  if (!m) return null;
+  const dateYmd = parseDayMonthYear(`${m[1]} ${m[2]} ${m[3]}`);
+  if (!dateYmd) return null;
+  const hh = String(Number(m[4])).padStart(2, "0");
+  const mm = m[5];
+  const ss = m[6] || "00";
+  const offset = m[7]
+    ? `${m[7].slice(0, 3)}:${m[7].slice(3)}`
+    : "+07:00";
+  return `${dateYmd}T${hh}:${mm}:${ss}${offset}`;
+}
+
+/**
+ * Parse time like "10:43:51 WIB" into HH:MM:SS.
+ */
+export function parseTimeWib(raw) {
+  if (!raw) return null;
+  const m = String(raw).trim().match(/(\d{1,2}):(\d{2})(?::(\d{2}))?/);
+  if (!m) return null;
+  const hh = String(Number(m[1])).padStart(2, "0");
+  const mm = m[2];
+  const ss = m[3] || "00";
+  return `${hh}:${mm}:${ss}`;
+}
+
+/** Combine YYYY-MM-DD + HH:MM:SS as WIB (+07:00) ISO string. */
+export function toWibIso(dateYmd, timeHms) {
+  if (!dateYmd) return null;
+  if (!timeHms) return dateYmd;
+  return `${dateYmd}T${timeHms}+07:00`;
+}
+
+/**
+ * Parse "10:47AM" / "10:47 AM" into 24h HH:MM:SS.
+ */
+export function parseAmPmTime(raw) {
+  if (!raw) return null;
+  const m = String(raw)
+    .trim()
+    .match(/^(\d{1,2}):(\d{2})(?::(\d{2}))?\s*([AaPp][Mm])$/);
+  if (!m) return null;
+  let hour = Number(m[1]);
+  const minute = m[2];
+  const second = m[3] || "00";
+  const ap = m[4].toUpperCase();
+  if (ap === "AM") {
+    if (hour === 12) hour = 0;
+  } else if (hour !== 12) {
+    hour += 12;
+  }
+  return `${String(hour).padStart(2, "0")}:${minute}:${second}`;
+}
+
+/**
+ * Capture value after a label, allowing optional whitespace between label and value
+ * (including jammed HTML-stripped forms like Tanggal13 Sep 2026).
+ */
+export function captureAfterLabel(body, label, { until } = {}) {
+  const escaped = label.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
+  const untilPart = until
+    ? `(?=${until.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")})`
+    : "(?=\\n|$)";
+  const re = new RegExp(`${escaped}\\s*([^\\n]*?)${untilPart}`, "i");
+  const m = body.match(re);
+  return m ? m[1].trim() : null;
+}
+
+/**
+ * Capture a multi-line block after a label until the next known label or blank run.
+ */
+export function captureBlockAfterLabel(body, label, nextLabels = []) {
+  const escaped = label.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
+  const next =
+    nextLabels.length > 0
+      ? nextLabels.map((l) => l.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")).join("|")
+      : null;
+  const until = next ? `(?=${next}|$)` : "$";
+  const re = new RegExp(`${escaped}\\s*([\\s\\S]*?)${until}`, "i");
+  const m = body.match(re);
+  if (!m) return null;
+  return m[1]
+    .split(/\r?\n/)
+    .map((l) => l.trim())
+    .filter(Boolean);
+}

apps/watcher/src/mime.js

@@ -0,0 +1,256 @@
+function decodeBase64Url(data) {
+  if (!data) return "";
+  const pad = data.length % 4 === 0 ? "" : "=".repeat(4 - (data.length % 4));
+  return Buffer.from(
+    data.replace(/-/g, "+").replace(/_/g, "/") + pad,
+    "base64",
+  ).toString("utf8");
+}
+
+function stripHtml(html) {
+  return html
+    .replace(/<style[\s\S]*?<\/style>/gi, "")
+    .replace(/<script[\s\S]*?<\/script>/gi, "")
+    .replace(/<br\s*\/?>/gi, "\n")
+    .replace(/<\/p>/gi, "\n\n")
+    .replace(/<\/div>/gi, "\n")
+    .replace(/<\/tr>/gi, "\n")
+    .replace(/<[^>]+>/g, "")
+    .replace(/&nbsp;/g, " ")
+    .replace(/&amp;/g, "&")
+    .replace(/&lt;/g, "<")
+    .replace(/&gt;/g, ">")
+    .replace(/&quot;/g, '"')
+    .replace(/&#39;/g, "'")
+    .replace(/&#x27;/gi, "'")
+    .replace(/\n{3,}/g, "\n\n")
+    .trim();
+}
+
+function walkParts(payload, acc = { text: "", html: "" }) {
+  if (!payload) return acc;
+  const mime = payload.mimeType || "";
+  const data = payload.body?.data;
+  if (data) {
+    if (mime === "text/plain") acc.text += decodeBase64Url(data);
+    else if (mime === "text/html") acc.html += decodeBase64Url(data);
+  }
+  for (const part of payload.parts || []) {
+    walkParts(part, acc);
+  }
+  return acc;
+}
+
+/** Return both MIME text and html parts (decoded). */
+export function extractMimeParts(payload) {
+  const { text, html } = walkParts(payload);
+  return {
+    text: text.trim(),
+    html: html.trim(),
+  };
+}
+
+export function extractBody(payload) {
+  const { text, html } = extractMimeParts(payload);
+  if (text) return text;
+  if (html) return stripHtml(html);
+  return "";
+}
+
+export function extractHtml(payload) {
+  return extractMimeParts(payload).html;
+}
+
+export function header(payload, name) {
+  const match = payload?.headers?.find(
+    (h) => h.name?.toLowerCase() === name.toLowerCase(),
+  );
+  return match?.value || "";
+}
+
+export function summarizeMessage(message) {
+  const payload = message.payload || {};
+  return {
+    id: message.id,
+    threadId: message.threadId,
+    from: header(payload, "From"),
+    to: header(payload, "To"),
+    subject: header(payload, "Subject") || "(no subject)",
+    date: header(payload, "Date"),
+    snippet: message.snippet || "",
+  };
+}
+
+function isRfc822Part(part) {
+  if (!part) return false;
+  const mime = (part.mimeType || "").toLowerCase();
+  const filename = (part.filename || "").toLowerCase();
+  return mime === "message/rfc822" || filename.endsWith(".eml");
+}
+
+/**
+ * Collect message/rfc822 (or *.eml) attachment parts. Does not recurse into
+ * nested .eml inside .eml — only direct attachment parts under the tree.
+ */
+export function listRfc822Parts(payload, acc = []) {
+  if (!payload) return acc;
+  for (const part of payload.parts || []) {
+    if (isRfc822Part(part)) {
+      acc.push(part);
+      continue; // do not recurse into nested eml
+    }
+    listRfc822Parts(part, acc);
+  }
+  return acc;
+}
+
+function findNestedHeaderSource(part) {
+  if (!part) return null;
+  if (header(part, "From") || header(part, "Subject")) return part;
+  for (const child of part.parts || []) {
+    const found = findNestedHeaderSource(child);
+    if (found) return found;
+  }
+  return null;
+}
+
+/**
+ * Build a synthetic message from a Gmail-expanded message/rfc822 part.
+ * When nested text/html is only available via attachmentId (common on large
+ * batches), call hydrateRfc822Message instead.
+ */
+export function messageFromRfc822Part(part, { parentMessageId, index = 0 } = {}) {
+  const headerSource = findNestedHeaderSource(part) || part;
+  const partId = part.partId || String(index);
+  const id = parentMessageId ? `${parentMessageId}#${partId}` : partId;
+  return {
+    id,
+    threadId: null,
+    from: header(headerSource, "From"),
+    to: header(headerSource, "To"),
+    subject: header(headerSource, "Subject") || "(no subject)",
+    date: header(headerSource, "Date"),
+    snippet: "",
+    body: extractBody(part),
+    html: extractHtml(part),
+    attachmentFilename: part.filename || "",
+    parentMessageId: parentMessageId || null,
+    _rfc822Part: part,
+  };
+}
+
+/** Collect text/plain + text/html nodes that need attachments.get. */
+export function listNestedBodyAttachmentRefs(part, acc = []) {
+  if (!part) return acc;
+  const mime = (part.mimeType || "").toLowerCase();
+  const attachId = part.body?.attachmentId;
+  if (attachId && (mime === "text/plain" || mime === "text/html") && !part.body?.data) {
+    acc.push({ mime, attachmentId: attachId });
+  }
+  for (const child of part.parts || []) {
+    listNestedBodyAttachmentRefs(child, acc);
+  }
+  return acc;
+}
+
+/**
+ * Decode a simple multipart .eml (no nested multipart recursion beyond one level
+ * of multipart/*) into { text, html }. Used when only the top-level rfc822 blob
+ * is downloadable.
+ */
+export function extractMimePartsFromRawEml(raw) {
+  if (!raw) return { text: "", html: "" };
+  let text = "";
+  let html = "";
+
+  // Split on MIME boundaries (best-effort)
+  const boundaryMatch = raw.match(/boundary="?([^";\r\n]+)"?/i);
+  const chunks = boundaryMatch
+    ? raw.split(new RegExp(`\\r?\\n--${boundaryMatch[1].replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}(?:--)?\\r?\\n`))
+    : [raw];
+
+  for (const chunk of chunks) {
+    const headerEnd = chunk.search(/\r?\n\r?\n/);
+    if (headerEnd < 0) continue;
+    const headers = chunk.slice(0, headerEnd);
+    let body = chunk.slice(headerEnd).replace(/^\r?\n\r?\n/, "");
+    const ct = headers.match(/Content-Type:\s*([^\r\n;]+)/i)?.[1]?.trim().toLowerCase() || "";
+    const cte = headers.match(/Content-Transfer-Encoding:\s*([^\r\n]+)/i)?.[1]?.trim().toLowerCase() || "";
+
+    if (cte === "base64") {
+      body = Buffer.from(body.replace(/\s+/g, ""), "base64").toString("utf8");
+    } else if (cte === "quoted-printable") {
+      body = body
+        .replace(/=\r?\n/g, "")
+        .replace(/=([0-9A-Fa-f]{2})/g, (_, h) => String.fromCharCode(parseInt(h, 16)));
+    }
+
+    if (ct === "text/plain") text += body;
+    else if (ct === "text/html") html += body;
+  }
+
+  return { text: text.trim(), html: html.trim() };
+}
+
+/**
+ * Ensure an rfc822-derived message has body/html. Fetches nested Gmail
+ * attachments when Gmail did not inline the content.
+ */
+export async function hydrateRfc822Message(gmail, parentMessageId, part, index = 0) {
+  const message = messageFromRfc822Part(part, { parentMessageId, index });
+  if (message.body || message.html) {
+    delete message._rfc822Part;
+    return message;
+  }
+
+  const refs = listNestedBodyAttachmentRefs(part);
+  let text = "";
+  let html = "";
+
+  for (const ref of refs) {
+    const { data } = await gmail.users.messages.attachments.get({
+      userId: "me",
+      messageId: parentMessageId,
+      id: ref.attachmentId,
+    });
+    const decoded = decodeBase64Url(data.data);
+    if (ref.mime === "text/plain") text += decoded;
+    else if (ref.mime === "text/html") html += decoded;
+  }
+
+  // Fallback: download the whole .eml blob
+  if (!text && !html && part.body?.attachmentId) {
+    const { data } = await gmail.users.messages.attachments.get({
+      userId: "me",
+      messageId: parentMessageId,
+      id: part.body.attachmentId,
+    });
+    const raw = decodeBase64Url(data.data);
+    const parts = extractMimePartsFromRawEml(raw);
+    text = parts.text;
+    html = parts.html;
+  }
+
+  message.html = html.trim();
+  message.body = text.trim() || (html ? stripHtml(html) : "");
+  delete message._rfc822Part;
+  return message;
+}
+
+/**
+ * Expand a fetched Gmail payload into work items:
+ * - if .eml / message/rfc822 attachments exist → those only
+ * - else → the outer message
+ */
+export function expandWorkItems(outerMessage, payload) {
+  const parts = listRfc822Parts(payload);
+  if (!parts.length) return [outerMessage];
+  return parts.map((part, index) =>
+    messageFromRfc822Part(part, {
+      parentMessageId: outerMessage.id,
+      index,
+    }),
+  );
+}
+
+

apps/watcher/src/notify/ntfy.js

@@ -0,0 +1,102 @@
+import { NTFY_URL } from "../config.js";
+
+/**
+ * POST a notification to ntfy. Never throws — failures are logged only.
+ */
+export async function notifyNtfy(
+  {
+    title,
+    message,
+    priority = "default",
+    tags = [],
+  },
+  log = console,
+) {
+  if (!NTFY_URL) return { skipped: true };
+
+  try {
+    const headers = {
+      Title: title.slice(0, 250),
+      Priority: String(priority),
+      "Content-Type": "text/plain; charset=utf-8",
+    };
+    if (tags.length) headers.Tags = tags.join(",");
+
+    const res = await fetch(NTFY_URL, {
+      method: "POST",
+      headers,
+      body: message,
+    });
+
+    if (!res.ok) {
+      const text = await res.text().catch(() => "");
+      log.warn?.(
+        { status: res.status, text: text.slice(0, 200) },
+        "ntfy: publish failed",
+      );
+      return { ok: false, status: res.status };
+    }
+    return { ok: true };
+  } catch (err) {
+    log.warn?.({ err: err.message || String(err) }, "ntfy: publish error");
+    return { ok: false, error: true };
+  }
+}
+
+/** Build a notification body for a failed/skipped receipt work item. */
+export function formatReceiptAlert({
+  reason,
+  messageId,
+  kind,
+  message,
+  record,
+  extra,
+}) {
+  const lines = [
+    `reason: ${reason}`,
+    `messageId: ${messageId || ""}`,
+    `kind: ${kind || ""}`,
+    `from: ${message?.from || ""}`,
+    `subject: ${message?.subject || ""}`,
+    `date: ${message?.date || ""}`,
+    `attachment: ${message?.attachmentFilename || ""}`,
+    `parentMessageId: ${message?.parentMessageId || ""}`,
+  ];
+
+  if (record) {
+    lines.push(`record: ${JSON.stringify(record)}`);
+  }
+  if (extra) {
+    lines.push(`extra: ${JSON.stringify(extra)}`);
+  }
+
+  const body = message?.body || "";
+  lines.push("", "----- message body -----", body || "(empty)");
+
+  if (message?.html) {
+    lines.push(
+      "",
+      "----- message html (length) -----",
+      String(message.html.length),
+    );
+  }
+
+  // ntfy payloads should stay reasonably sized
+  const text = lines.join("\n");
+  const max = 12000;
+  if (text.length <= max) return text;
+  return `${text.slice(0, max)}\n\n… truncated (${text.length} chars total)`;
+}
+
+export async function notifyReceiptIssue(payload, log) {
+  const body = formatReceiptAlert(payload);
+  return notifyNtfy(
+    {
+      title: `gmail-hook: ${payload.reason}`,
+      message: body,
+      priority: "high",
+      tags: ["warning", "email"],
+    },
+    log,
+  );
+}

apps/watcher/src/parsers/bri.js

@@ -0,0 +1,68 @@
+import {
+  captureAfterLabel,
+  captureBlockAfterLabel,
+  parseDayMonthYear,
+  parseIdrAmount,
+  parseTimeWib,
+  toWibIso,
+} from "../lib/parse-helpers.js";
+
+/**
+ * Parse BRImo / Bank BRI transfer success body (plain text).
+ * Required: Nominal + Nomor Referensi.
+ */
+export function parseBriTransfer(body) {
+  if (!body) return null;
+
+  const payerMatch = body.match(/Halo[,\s*]*([^*\n,]+)/i);
+  const payerName = payerMatch ? payerMatch[1].replace(/\*/g, "").trim() : null;
+
+  const tanggalRaw = captureAfterLabel(body, "Tanggal");
+  const dateYmd = parseDayMonthYear(tanggalRaw);
+  const timeHms = parseTimeWib(tanggalRaw);
+  const occurredAt = toWibIso(dateYmd, timeHms);
+
+  // Prefer Nominal over Total (Total = Nominal + Biaya Admin).
+  const nominalRaw = captureAfterLabel(body, "Nominal");
+  const amount = parseIdrAmount(nominalRaw);
+
+  const reference =
+    captureAfterLabel(body, "Nomor Referensi") ||
+    body.match(/Nomor\s+Referensi\s*([^\s\n]+)/i)?.[1] ||
+    null;
+
+  if (amount == null || !reference) return null;
+
+  const namaTujuan = captureAfterLabel(body, "Nama Tujuan");
+  const jenisTransaksi = captureAfterLabel(body, "Jenis Transaksi");
+  const bankTujuan = captureAfterLabel(body, "Bank Tujuan");
+  const nomorTujuan = captureAfterLabel(body, "Nomor Tujuan");
+  const catatanRaw = captureAfterLabel(body, "Catatan");
+  const biayaAdmin = parseIdrAmount(captureAfterLabel(body, "Biaya Admin"));
+
+  const fundingLines = captureBlockAfterLabel(body, "Sumber Dana", [
+    "Jenis Transaksi",
+  ]);
+  const fundingSource =
+    fundingLines?.join(" ").replace(/\s+/g, " ").trim() || null;
+
+  const details = {};
+  if (payerName) details.payerName = payerName;
+  if (namaTujuan) details.namaTujuan = namaTujuan.trim();
+  if (jenisTransaksi) details.jenisTransaksi = jenisTransaksi.trim();
+  if (bankTujuan) details.bankTujuan = bankTujuan.trim();
+  if (nomorTujuan) details.nomorTujuan = nomorTujuan.trim();
+  if (catatanRaw != null && catatanRaw.trim() && catatanRaw.trim() !== "-") {
+    details.catatan = catatanRaw.trim();
+  }
+  if (biayaAdmin != null) details.biayaAdmin = biayaAdmin;
+  if (fundingSource) details.fundingSource = fundingSource;
+
+  return {
+    amount,
+    currency: "IDR",
+    reference: reference.trim(),
+    occurredAt,
+    details,
+  };
+}

apps/watcher/src/parsers/grab.js

@@ -0,0 +1,247 @@
+import { parse } from "node-html-parser";
+import {
+  captureAfterLabel,
+  captureBlockAfterLabel,
+  parseAmPmTime,
+  parseDayMonthYear,
+  parseGrabDateTime,
+  parseIdrAmount,
+  toWibIso,
+} from "../lib/parse-helpers.js";
+
+function stripForwardedHeader(body) {
+  const m = body.match(
+    /-{2,}\s*Forwarded message\s*-{2,}[\s\S]*?\nTo:[^\n]*\n+/i,
+  );
+  if (!m) return body;
+  return body.slice(m.index + m[0].length);
+}
+
+/** Tip receipts share From/Subject with ride e-receipts but different body copy. */
+export function isGrabTipBody(body) {
+  if (!body) return false;
+  if (/your tip goes a long way/i.test(body)) return true;
+  if (/thanks!\s*your tip/i.test(body)) return true;
+  // Tip layout: has a Tip section, no ride pickup line
+  if (
+    !/picked up on/i.test(body) &&
+    !/hope you enjoyed your ride/i.test(body) &&
+    !/selamat menikmati makanan/i.test(body) &&
+    !/grabfood/i.test(body) &&
+    /\btip\b/i.test(body) &&
+    /total paid/i.test(body)
+  ) {
+    return true;
+  }
+  return false;
+}
+
+/** GrabFood order receipts (ID/EN) — same From/Subject as rides. */
+export function isGrabFoodBody(body) {
+  if (!body) return false;
+  if (/selamat menikmati makanan/i.test(body)) return true;
+  if (/jenis kendaraan:\s*\r?\n\s*GrabFood/i.test(body)) return true;
+  if (/\bGrabFood\b/i.test(body) && /pesanan id/i.test(body)) return true;
+  return false;
+}
+
+function extractProduct(content) {
+  const productMatch = content.match(
+    /^\s*(Bike\s+\w+|Car\s+\w+|Grab\w*|Food|Express|Mart)[^\n]*/im,
+  );
+  return productMatch ? productMatch[1].trim() : null;
+}
+
+function extractBookingId(body) {
+  return (
+    captureAfterLabel(body, "Booking ID:") ||
+    body.match(/Booking ID:\s*([A-Z0-9-]+)/i)?.[1] ||
+    null
+  );
+}
+
+function extractTotalPaid(body) {
+  return (
+    captureAfterLabel(body, "Total Paid") ||
+    body.match(/Total Paid\s*\r?\n\s*(Rp[\d.,]+)/i)?.[1]
+  );
+}
+
+/**
+ * Grab ride HTML puts pickup/dropoff as standalone "10:47AM" divs next to
+ * locations. Plain text only has "Picked up on 13 September 2026" (date only).
+ * Returns { pickup, dropoff } as HH:MM:SS (24h) when found.
+ */
+export function extractGrabTripTimesFromHtml(html) {
+  if (!html) return { pickup: null, dropoff: null };
+  const root = parse(html);
+  const times = [];
+  for (const el of root.querySelectorAll("div")) {
+    const t = (el.text || "").replace(/\u00a0/g, " ").trim();
+    if (!/^\d{1,2}:\d{2}(?::\d{2})?\s*[AaPp][Mm]$/.test(t)) continue;
+    const hms = parseAmPmTime(t);
+    if (hms) times.push(hms);
+  }
+  return {
+    pickup: times[0] || null,
+    dropoff: times[1] || null,
+  };
+}
+
+function parseRide(body, content, html) {
+  const product = extractProduct(content);
+  const pickedUpRaw = captureAfterLabel(body, "Picked up on");
+  const dateYmd = parseDayMonthYear(pickedUpRaw);
+  const { pickup, dropoff } = extractGrabTripTimesFromHtml(html);
+  const occurredAt = toWibIso(dateYmd, pickup) || dateYmd;
+
+  const bookingId = extractBookingId(body);
+  const amount = parseIdrAmount(extractTotalPaid(body));
+
+  if (amount == null || !bookingId) return null;
+
+  const details = { variant: "ride" };
+  if (product) details.product = product;
+  if (dropoff) details.dropoffAt = toWibIso(dateYmd, dropoff);
+
+  const driverMatch = body.match(
+    /Receipt issued by the driver\s+([^.\n]+)/i,
+  );
+  if (driverMatch) details.driver = driverMatch[1].trim();
+
+  const ratingMatch = body.match(
+    /(\d(?:\.\d)?)\s*(?:\[image: stars\]|★|stars)/i,
+  );
+  if (ratingMatch) details.driverRating = Number(ratingMatch[1]);
+
+  const complimentsBlock = body.match(
+    /Compliments for driver\s*\r?\n([\s\S]*?)(?=\r?\nBreakdown|\r?\nPassenger|$)/i,
+  );
+  if (complimentsBlock) {
+    const compliments = complimentsBlock[1]
+      .split(/\r?\n/)
+      .map((l) => l.trim())
+      .filter((l) => l && !/^\[image/i.test(l));
+    if (compliments.length) details.compliments = compliments;
+  }
+
+  const passenger = captureAfterLabel(body, "Passenger");
+  if (passenger) {
+    details.passenger = passenger.replace(/\s*Profile.*$/i, "").trim();
+  }
+
+  return {
+    variant: "ride",
+    kind: "grab.ereceipt.ride",
+    amount,
+    currency: "IDR",
+    reference: bookingId.trim(),
+    occurredAt,
+    details,
+  };
+}
+
+function parseTip(body, content) {
+  const product = extractProduct(content);
+  const bookingId = extractBookingId(body);
+  const amount = parseIdrAmount(extractTotalPaid(body));
+
+  if (amount == null || !bookingId) return null;
+
+  // Tip timestamp sits on its own line, e.g. "13 Sep 26 10:47 +0700"
+  const tipTs =
+    body.match(
+      /(\d{1,2}\s+[A-Za-z]+\s+\d{2,4}\s+\d{1,2}:\d{2}(?::\d{2})?\s*[+-]\d{4})/,
+    )?.[1] || null;
+  const occurredAt = parseGrabDateTime(tipTs);
+
+  const details = { variant: "tip" };
+  if (product) details.product = product;
+
+  return {
+    variant: "tip",
+    kind: "grab.ereceipt.tip",
+    amount,
+    currency: "IDR",
+    // Separate invoice from the ride spend for the same Booking ID
+    reference: `${bookingId.trim()}:tip`,
+    occurredAt,
+    details,
+  };
+}
+
+function parseFood(body, content) {
+  const orderId =
+    captureAfterLabel(body, "Pesanan ID") ||
+    body.match(/Pesanan ID\s*\r?\n\s*([A-Z0-9-]+)/i)?.[1] ||
+    captureAfterLabel(body, "Order ID") ||
+    null;
+
+  const totalRaw =
+    body.match(/TOTAL\s*\(INCL\.?\s*TAX\)\s*IDR\s*([\d.]+)/i)?.[0] ||
+    body.match(/TOTAL\s*\r?\n\s*IDR\s*([\d.]+)/i)?.[0] ||
+    body.match(/IDR\s*[\d.]+/)?.[0];
+  const amount = parseIdrAmount(totalRaw);
+
+  if (amount == null || !orderId) return null;
+
+  const occurredAt = parseGrabDateTime(
+    body.match(
+      /(\d{1,2}\s+[A-Za-z]+\s+\d{2,4}\s+\d{1,2}:\d{2}(?::\d{2})?\s*[+-]\d{4})/,
+    )?.[1],
+  );
+
+  const details = { variant: "food", product: "Food" };
+
+  const vehicle =
+    captureAfterLabel(body, "Jenis Kendaraan:") ||
+    body.match(/Jenis Kendaraan:\s*\r?\n\s*([^\n]+)/i)?.[1];
+  if (vehicle) details.vehicleType = vehicle.trim();
+
+  const forName =
+    captureAfterLabel(body, "Diterbitkan untuk") ||
+    body.match(/Diterbitkan untuk\s*\r?\n\s*([^\n]+)/i)?.[1];
+  if (forName) details.passenger = forName.trim();
+
+  const merchantLines = captureBlockAfterLabel(body, "Pesanan Dari:", [
+    "Profil:",
+    "Detail Tagihan",
+    "Detail Pembayaran",
+  ]);
+  if (merchantLines?.length) {
+    details.merchant = merchantLines[0];
+    if (merchantLines.length > 1) {
+      details.location = merchantLines
+        .slice(1)
+        .join(" ")
+        .replace(/<https?:\/\/[^>]+>/g, "")
+        .replace(/\s+/g, " ")
+        .trim();
+    }
+  }
+
+  const payment =
+    captureAfterLabel(body, "Detail Pembayaran:") ||
+    body.match(/Detail Pembayaran:\s*\r?\n\s*([^\n]+)/i)?.[1];
+  if (payment && !/Jumlah/i.test(payment)) {
+    details.paymentMethod = payment.trim();
+  }
+
+  return {
+    variant: "food",
+    kind: "grab.ereceipt.food",
+    amount,
+    currency: "IDR",
+    reference: orderId.trim(),
+    occurredAt,
+    details,
+  };
+}
+
+export function parseGrabReceipt(body, { html = "" } = {}) {
+  if (!body) return null;
+  const content = stripForwardedHeader(body);
+  if (isGrabTipBody(body)) return parseTip(body, content);
+  if (isGrabFoodBody(body)) return parseFood(body, content);
+  return parseRide(body, content, html);
+}

apps/watcher/src/parsers/livin.js

@@ -0,0 +1,94 @@
+import {
+  captureAfterLabel,
+  captureBlockAfterLabel,
+  parseDayMonthYear,
+  parseIdrAmount,
+  parseTimeWib,
+  toWibIso,
+} from "../lib/parse-helpers.js";
+
+/**
+ * Split Penerima lines into merchant + location when the body already
+ * separates them. If HTML-stripped glue merges them into one line
+ * (e.g. SNACK BAROKAHJAKARTA BARAT - ID), keep the whole string as merchant
+ * — do not guess Indonesian city names.
+ */
+export function splitMerchantLocation(lines) {
+  if (!lines || !lines.length) {
+    return { merchant: null, location: null };
+  }
+
+  if (lines.length >= 2) {
+    const last = lines[lines.length - 1];
+    if (/-\s*ID\s*$/i.test(last)) {
+      return {
+        merchant: lines.slice(0, -1).join(" ").trim() || null,
+        location: last.trim(),
+      };
+    }
+  }
+
+  return { merchant: lines.join(" ").trim() || null, location: null };
+}
+
+export function parseLivinReceipt(body) {
+  if (!body) return null;
+
+  const payerMatch = body.match(/Halo\s+([^,\n]+)/i);
+  const payerName = payerMatch ? payerMatch[1].trim() : null;
+
+  const penerimaLines = captureBlockAfterLabel(body, "Penerima", [
+    "Tanggal",
+    "Jam",
+    "Nominal",
+  ]);
+  const { merchant, location } = splitMerchantLocation(penerimaLines);
+
+  const tanggalRaw = captureAfterLabel(body, "Tanggal");
+  const jamRaw = captureAfterLabel(body, "Jam");
+  const dateYmd = parseDayMonthYear(tanggalRaw);
+  const timeHms = parseTimeWib(jamRaw);
+  const occurredAt = toWibIso(dateYmd, timeHms);
+
+  const nominalRaw = captureAfterLabel(body, "Nominal Transaksi");
+  const amount = parseIdrAmount(nominalRaw);
+
+  const reference =
+    captureAfterLabel(body, "No. Referensi") ||
+    body.match(/No\.\s*Referensi\s*([^\s\n]+)/i)?.[1] ||
+    null;
+
+  if (amount == null || !reference) return null;
+
+  const details = {};
+  if (payerName) details.payerName = payerName;
+  if (merchant) details.merchant = merchant;
+  if (location) details.location = location;
+
+  const qrisRef =
+    captureAfterLabel(body, "No. Ref. QRIS") ||
+    body.match(/No\.\s*Ref\.\s*QRIS\s*([^\s\n]+)/i)?.[1];
+  const merchantPan = captureAfterLabel(body, "Merchant PAN");
+  const customerPan = captureAfterLabel(body, "Customer PAN");
+  const acquirer = captureAfterLabel(body, "Pengakuisisi");
+  const terminalId = captureAfterLabel(body, "Terminal ID");
+  const fundingLines = captureBlockAfterLabel(body, "Sumber Dana", [
+    "Simpan email",
+  ]);
+  const fundingSource = fundingLines?.join(" ").replace(/\s+/g, " ").trim() || null;
+
+  if (qrisRef) details.qrisRef = qrisRef;
+  if (merchantPan) details.merchantPan = merchantPan;
+  if (customerPan) details.customerPan = customerPan;
+  if (acquirer) details.acquirer = acquirer;
+  if (terminalId) details.terminalId = terminalId.trim();
+  if (fundingSource) details.fundingSource = fundingSource;
+
+  return {
+    amount,
+    currency: "IDR",
+    reference: reference.trim(),
+    occurredAt,
+    details,
+  };
+}

apps/watcher/src/pipeline.js

@@ -0,0 +1,12 @@
+export async function runPipeline(handlers, ctx) {
+  let current = { ...ctx };
+  for (const handler of handlers) {
+    const result = await handler(current);
+    if (!result) continue;
+    if (result.pass && typeof result.pass === "object" && !Array.isArray(result.pass)) {
+      current = { ...current, ...result.pass };
+    }
+    if (result.break) break;
+  }
+  return current;
+}

apps/watcher/src/pocketbase/client.js

@@ -0,0 +1,91 @@
+import {
+  POCKETBASE_PASSWORD,
+  POCKETBASE_URL,
+  POCKETBASE_USER,
+} from "../config.js";
+
+let cachedToken = null;
+let cachedTokenAt = 0;
+const TOKEN_TTL_MS = 50 * 60 * 1000; // refresh before typical 1h expiry
+
+export function pocketbaseConfigured() {
+  return Boolean(POCKETBASE_URL && POCKETBASE_USER && POCKETBASE_PASSWORD);
+}
+
+async function authSuperuser() {
+  const res = await fetch(
+    `${POCKETBASE_URL}/api/collections/_superusers/auth-with-password`,
+    {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        identity: POCKETBASE_USER,
+        password: POCKETBASE_PASSWORD,
+      }),
+    },
+  );
+  const data = await res.json().catch(() => ({}));
+  if (!res.ok || !data.token) {
+    throw new Error(
+      `PocketBase auth failed (${res.status}): ${data.message || "unknown"}`,
+    );
+  }
+  cachedToken = data.token;
+  cachedTokenAt = Date.now();
+  return cachedToken;
+}
+
+async function getToken() {
+  if (cachedToken && Date.now() - cachedTokenAt < TOKEN_TTL_MS) {
+    return cachedToken;
+  }
+  return authSuperuser();
+}
+
+export async function pbRequest(path, { method = "GET", body, retry = true } = {}) {
+  if (!pocketbaseConfigured()) {
+    throw new Error("PocketBase is not configured");
+  }
+  const token = await getToken();
+  const res = await fetch(`${POCKETBASE_URL}${path}`, {
+    method,
+    headers: {
+      Authorization: token,
+      ...(body ? { "Content-Type": "application/json" } : {}),
+    },
+    body: body ? JSON.stringify(body) : undefined,
+  });
+
+  if (res.status === 401 && retry) {
+    cachedToken = null;
+    await authSuperuser();
+    return pbRequest(path, { method, body, retry: false });
+  }
+
+  const text = await res.text();
+  let data = {};
+  try {
+    data = text ? JSON.parse(text) : {};
+  } catch {
+    data = { raw: text };
+  }
+  return { ok: res.ok, status: res.status, data };
+}
+
+export async function getCollection(name) {
+  return pbRequest(`/api/collections/${name}`);
+}
+
+export async function updateCollection(name, patch) {
+  return pbRequest(`/api/collections/${name}`, {
+    method: "PATCH",
+    body: patch,
+  });
+}
+
+export async function createRecord(collection, record) {
+  return pbRequest(`/api/collections/${collection}/records`, {
+    method: "POST",
+    body: record,
+  });
+}

apps/watcher/src/pocketbase/map.js

@@ -0,0 +1,52 @@
+/**
+ * Map parsed receipt envelope → PocketBase spendings record.
+ */
+export function toSpendingRecord(parsed, rule) {
+  const source = rule?.source || parsed.kind?.split(".")[0] || "";
+  let description = "";
+  if (source === "livin") {
+    description = parsed.details?.merchant || "";
+  } else if (source === "bri") {
+    description = parsed.details?.namaTujuan || "";
+  } else if (source === "grab") {
+    if (parsed.kind === "grab.ereceipt.food" || parsed.details?.variant === "food") {
+      const merchant = parsed.details?.merchant;
+      description = merchant ? `Grab Food ${merchant}` : "Grab Food";
+    } else {
+      const product = parsed.details?.product;
+      const base = product ? `Grab ${product}` : "Grab";
+      description =
+        parsed.kind === "grab.ereceipt.tip" || parsed.details?.variant === "tip"
+          ? `${base} tip`
+          : base;
+    }
+  }
+
+  return {
+    invoice_id: parsed.reference,
+    amount: parsed.amount,
+    trx_date: toPocketBaseDate(parsed.occurredAt),
+    source,
+    description,
+    category: "",
+    message_id: parsed.messageId || "",
+    details: parsed.details || {},
+    // Full watcher envelope snapshot for audit / debugging (not the raw email body).
+    parsed,
+  };
+}
+
+/**
+ * PocketBase date fields accept "YYYY-MM-DD HH:mm:ss.SSSZ" (UTC) or date-only.
+ */
+export function toPocketBaseDate(occurredAt) {
+  if (!occurredAt) return null;
+  // Date-only (Grab)
+  if (/^\d{4}-\d{2}-\d{2}$/.test(occurredAt)) {
+    return `${occurredAt} 00:00:00.000Z`;
+  }
+  const d = new Date(occurredAt);
+  if (Number.isNaN(d.getTime())) return null;
+  // PocketBase: "2026-09-13 07:48:31.000Z"
+  return d.toISOString().replace("T", " ");
+}

apps/watcher/src/pocketbase/schema.js

@@ -0,0 +1,121 @@
+import { getCollection, pocketbaseConfigured, updateCollection } from "./client.js";
+
+const SPENDINGS = "spendings";
+
+const DESIRED_FIELDS = [
+  {
+    name: "invoice_id",
+    type: "text",
+    required: true,
+    min: 0,
+  },
+  {
+    name: "amount",
+    type: "number",
+    required: true,
+  },
+  {
+    name: "trx_date",
+    type: "date",
+    required: true,
+  },
+  {
+    name: "source",
+    type: "text",
+    required: true,
+    min: 0,
+  },
+  {
+    name: "description",
+    type: "text",
+    required: false,
+    min: 0,
+  },
+  {
+    name: "category",
+    type: "text",
+    required: false,
+    min: 0,
+  },
+  {
+    name: "message_id",
+    type: "text",
+    required: false,
+    min: 0,
+  },
+  {
+    name: "details",
+    type: "json",
+    required: false,
+  },
+  {
+    name: "parsed",
+    type: "json",
+    required: false,
+  },
+];
+
+const UNIQUE_INVOICE_INDEX =
+  "CREATE UNIQUE INDEX IF NOT EXISTS idx_spendings_invoice_id ON spendings (invoice_id)";
+
+/**
+ * Ensure spendings has the fields and unique invoice_id index we need.
+ * Safe on an empty collection. Idempotent.
+ */
+export async function ensureSpendingsSchema(log = console) {
+  if (!pocketbaseConfigured()) {
+    log.warn?.("pocketbase: skipped schema ensure (not configured)");
+    return { skipped: true };
+  }
+
+  const { ok, status, data } = await getCollection(SPENDINGS);
+  if (!ok) {
+    throw new Error(
+      `pocketbase: cannot load spendings collection (${status}): ${data.message || ""}`,
+    );
+  }
+
+  const existing = data.fields || [];
+  const byName = new Map(existing.map((f) => [f.name, f]));
+
+  // Keep system/autodate fields; merge desired custom fields by name.
+  const systemNames = new Set(["id", "created", "updated"]);
+  const nextFields = existing.filter((f) => systemNames.has(f.name));
+
+  for (const desired of DESIRED_FIELDS) {
+    const prev = byName.get(desired.name);
+    if (prev) {
+      nextFields.push({
+        ...prev,
+        type: desired.type,
+        required: desired.required,
+        ...(desired.min != null ? { min: desired.min } : {}),
+      });
+    } else {
+      nextFields.push({ ...desired });
+    }
+  }
+
+  // Preserve any other unknown custom fields the user may have added.
+  for (const f of existing) {
+    if (systemNames.has(f.name)) continue;
+    if (DESIRED_FIELDS.some((d) => d.name === f.name)) continue;
+    nextFields.push(f);
+  }
+
+  const indexes = Array.isArray(data.indexes) ? [...data.indexes] : [];
+  if (!indexes.some((idx) => /invoice_id/i.test(String(idx)))) {
+    indexes.push(UNIQUE_INVOICE_INDEX);
+  }
+
+  const patch = { fields: nextFields, indexes };
+  const result = await updateCollection(SPENDINGS, patch);
+  if (!result.ok) {
+    throw new Error(
+      `pocketbase: failed to update spendings schema (${result.status}): ${result.data.message || JSON.stringify(result.data)}`,
+    );
+  }
+
+  log.info?.({ collection: SPENDINGS }, "pocketbase: spendings schema ensured");
+  return { ok: true };
+}

apps/watcher/src/process-fetched.js

@@ -0,0 +1,32 @@
+import { runPipeline } from "./pipeline.js";
+
+/**
+ * After fetch: if .eml attachments exist, process each; otherwise process outer.
+ * `fetchResult` is the ctx after [logEvent, fetchAndLogMessage].
+ */
+export async function processFetchedMessage(fetchResult, processHandlers) {
+  if (!fetchResult?.message) return [];
+
+  const workItems = fetchResult.emlMessages?.length
+    ? fetchResult.emlMessages
+    : [fetchResult.message];
+
+  const results = [];
+  for (const item of workItems) {
+    const result = await runPipeline(processHandlers, {
+      ...fetchResult,
+      message: item,
+      messageId: item.id,
+      // Clear prior match/parse/save state from a previous item
+      rule: undefined,
+      original: undefined,
+      parsed: undefined,
+      saved: undefined,
+      duplicate: undefined,
+      error: undefined,
+      spendingId: undefined,
+    });
+    results.push(result);
+  }
+  return results;
+}

apps/watcher/src/routes/pubsub.js

@@ -0,0 +1,169 @@
+import { timingSafeEqual } from "node:crypto";
+import { PUBSUB_VERIFICATION_TOKEN } from "../config.js";
+import { getUserByEmail, getWatchState, saveWatchState } from "../db.js";
+import {
+  gmailClientForUser,
+  googleStatus,
+  listHistory,
+  startWatch,
+} from "../google.js";
+import { HttpError } from "../http.js";
+import { runPipeline } from "../pipeline.js";
+import { processFetchedMessage } from "../process-fetched.js";
+
+const userLocks = new Map();
+
+function withUserLock(userId, fn) {
+  const prev = userLocks.get(userId) || Promise.resolve();
+  const next = prev.then(fn, fn);
+  userLocks.set(
+    userId,
+    next.then(
+      () => {},
+      () => {},
+    ),
+  );
+  return next;
+}
+
+function tokenFromRequest(request) {
+  const q = request.query?.token;
+  if (typeof q === "string" && q) return q;
+  const header =
+    request.headers["x-pubsub-verification-token"] ||
+    request.headers["x-goog-pubsub-verification-token"];
+  if (typeof header === "string" && header) return header;
+  const auth = request.headers.authorization;
+  if (typeof auth === "string" && auth.startsWith("Bearer ")) {
+    return auth.slice(7);
+  }
+  return "";
+}
+
+function tokensMatch(provided, expected) {
+  const a = Buffer.from(provided);
+  const b = Buffer.from(expected);
+  if (a.length !== b.length) return false;
+  return timingSafeEqual(a, b);
+}
+
+function verifyPush(request) {
+  if (!PUBSUB_VERIFICATION_TOKEN) return;
+  if (!tokensMatch(tokenFromRequest(request), PUBSUB_VERIFICATION_TOKEN)) {
+    throw new HttpError(403, "Invalid Pub/Sub verification token");
+  }
+}
+
+function decodeNotification(body) {
+  const data = body?.message?.data;
+  if (typeof data !== "string" || !data) return null;
+  try {
+    const json = Buffer.from(data, "base64").toString("utf8");
+    const parsed = JSON.parse(json);
+    if (!parsed?.emailAddress || parsed.historyId == null) return null;
+    return {
+      emailAddress: parsed.emailAddress,
+      historyId: String(parsed.historyId),
+    };
+  } catch {
+    return null;
+  }
+}
+
+async function processNotification(log, { fetchHandlers, processHandlers }, notification) {
+  const user = getUserByEmail(notification.emailAddress);
+  if (!user) {
+    log.warn(
+      { email: notification.emailAddress },
+      "gmail-watch: unknown mailbox, acking",
+    );
+    return;
+  }
+
+  await withUserLock(user.id, () =>
+    handleUserNotification(log, { fetchHandlers, processHandlers }, user, notification),
+  );
+}
+
+async function handleUserNotification(
+  log,
+  { fetchHandlers, processHandlers },
+  user,
+  notification,
+) {
+  const gmail = await gmailClientForUser(user.id);
+
+  const state = getWatchState(user.id);
+  const startHistoryId = state?.history_id;
+  if (!startHistoryId) {
+    log.info(
+      { userId: user.id },
+      "gmail-watch: no cursor yet, starting watch and skipping backlog",
+    );
+    await startWatch(user.id, { resetCursor: true });
+    return;
+  }
+
+  let result;
+  try {
+    result = await listHistory(gmail, startHistoryId);
+  } catch (err) {
+    const status = googleStatus(err);
+    if (status === 404) {
+      log.warn(
+        { userId: user.id, startHistoryId },
+        "gmail-watch: history expired, re-watching without backfill",
+      );
+      await startWatch(user.id, { resetCursor: true });
+      return;
+    }
+    throw err;
+  }
+
+  for (const messageId of result.messageIds) {
+    const fetched = await runPipeline(fetchHandlers, {
+      notification,
+      userId: user.id,
+      messageId,
+      gmail,
+      log,
+    });
+    await processFetchedMessage(fetched, processHandlers);
+  }
+
+  saveWatchState(user.id, {
+    historyId: result.historyId || notification.historyId,
+    expiration: state.expiration || Date.now() + 7 * 24 * 3600 * 1000,
+  });
+}
+
+export async function registerPubsubRoutes(app, { fetchHandlers, processHandlers }) {
+  app.post("/pubsub/gmail", async (request, reply) => {
+    verifyPush(request);
+
+    const notification = decodeNotification(request.body);
+    if (!notification) {
+      request.log.warn("gmail-watch: undecodable Pub/Sub payload, acking");
+      return reply.code(204).send();
+    }
+
+    try {
+      await processNotification(
+        request.log,
+        { fetchHandlers, processHandlers },
+        notification,
+      );
+    } catch (err) {
+      const status = googleStatus(err) || err.statusCode || 500;
+      if (status === 401 || status === 403) {
+        throw new HttpError(
+          500,
+          "Gmail access expired or was revoked. Sign in again.",
+        );
+      }
+      throw err;
+    }
+
+    return reply.code(204).send();
+  });
+}

apps/watcher/src/rules/bri-transfer.js

@@ -0,0 +1,6 @@
+export const briTransfer = {
+  kind: "bri.transfer.success",
+  source: "bri",
+  fromAddress: "bankbri@bri.co.id",
+  subject: "Pemindahan Dana Sesama Rekening BRI",
+};

apps/watcher/src/rules/grab-ereceipt.js

@@ -0,0 +1,6 @@
+export const grabEreceipt = {
+  kind: "grab.ereceipt.ride",
+  source: "grab",
+  fromAddress: "no-reply@grab.com",
+  subject: "Your Grab E-Receipt",
+};

apps/watcher/src/rules/index.js

@@ -0,0 +1,5 @@
+import { briTransfer } from "./bri-transfer.js";
+import { grabEreceipt } from "./grab-ereceipt.js";
+import { livinQrPayment } from "./livin-qr-payment.js";
+
+export const rules = [livinQrPayment, grabEreceipt, briTransfer];

apps/watcher/src/rules/livin-qr-payment.js

@@ -0,0 +1,6 @@
+export const livinQrPayment = {
+  kind: "livin.qr_payment.success",
+  source: "livin",
+  fromAddress: "noreply.livin@bankmandiri.co.id",
+  subject: "Pembayaran Berhasil!",
+};