Repository navigation
feat(usage-hub): charges auto-matching wiring — producer export, collector matcher, once-only surfacing (#1057) #2182
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| #!/usr/bin/env bun | ||
| // Charges producer (#1057): the Muse-side half of the charges | ||
| // auto-matcher. The aggregator's collector cannot reach the Plaid or | ||
| // Gmail connectors in-process, so an agent holding those connectors | ||
| // saves their raw reads to files and this script normalizes them into | ||
| // the charges export the collector ingests | ||
| // (~/atlas/charges-export/latest.json — the file-export convention, | ||
| // gather-quota.ts family 2). | ||
| // | ||
| // Inputs (both fail-soft; either may be absent): | ||
| // PLAID_TRANSACTIONS_PATH — a Plaid transactions-get response saved | ||
| // verbatim: { accounts?: [{account_id, name, mask}], transactions: | ||
| // [{date, name, merchant_name?, amount, pending?}] }. Plaid's | ||
| // amount sign: positive = money out. Pending rows are skipped. | ||
| // Balances are never read or written — charge evidence only. | ||
| // GMAIL_ALERTS_PATH — receipt/card alert extractions as an array of | ||
| // {at, merchant, amountUsd, method?} (the reading agent extracts | ||
| // these fields from the mail; this script does not parse mail). | ||
| // | ||
| // Output: { generatedAt, candidates: ChargeCandidate[] } written | ||
| // atomically to CHARGES_EXPORT_PATH (default | ||
| // ~/atlas/charges-export/latest.json). The collector's matcher | ||
| // dedupes across sources, so overlapping reads are safe; re-running | ||
| // for the same period republishes the same candidates. | ||
| // | ||
| // Usage: bun run scripts/gather-charges.ts | ||
| import { mkdir, readFile, rename, writeFile } from "node:fs/promises"; | ||
| import { homedir } from "node:os"; | ||
| import { dirname, join } from "node:path"; | ||
| import type { ChargeCandidate } from "../src/charges-match.ts"; | ||
| import { parseChargeCandidatesExport } from "../src/charges-collect.ts"; | ||
|
|
||
| const HOME = homedir(); | ||
| const plaidPath = process.env.PLAID_TRANSACTIONS_PATH ?? join(HOME, "atlas/charges-export/plaid-transactions.json"); | ||
| const gmailPath = process.env.GMAIL_ALERTS_PATH ?? join(HOME, "atlas/charges-export/gmail-alerts.json"); | ||
| const outPath = process.env.CHARGES_EXPORT_PATH ?? join(HOME, "atlas/charges-export/latest.json"); | ||
|
|
||
| interface PlaidAccount { | ||
| account_id?: string; | ||
| name?: string; | ||
| mask?: string; | ||
| } | ||
| interface PlaidTransaction { | ||
| date?: string; | ||
| name?: string; | ||
| merchant_name?: string; | ||
| amount?: number; | ||
| pending?: boolean; | ||
| account_id?: string; | ||
| } | ||
|
|
||
| function plaidCandidates(doc: unknown): ChargeCandidate[] { | ||
| const d = doc as { accounts?: PlaidAccount[]; transactions?: PlaidTransaction[] }; | ||
| if (!Array.isArray(d?.transactions)) return []; | ||
| const accounts = new Map((d.accounts ?? []).map((a) => [a.account_id, a])); | ||
| const out: ChargeCandidate[] = []; | ||
| for (const tx of d.transactions) { | ||
| if (tx.pending) continue; | ||
| if (typeof tx.amount !== "number" || !(tx.amount > 0)) continue; | ||
| if (typeof tx.date !== "string") continue; | ||
| const merchant = (tx.merchant_name || tx.name || "").trim(); | ||
| if (!merchant) continue; | ||
| const account = tx.account_id ? accounts.get(tx.account_id) : undefined; | ||
| const method = | ||
| account?.name && account?.mask | ||
| ? account.name + " •" + account.mask | ||
| : account?.mask | ||
| ? "•" + account.mask | ||
| : undefined; | ||
| out.push({ source: "plaid", at: tx.date, merchant, amountUsd: tx.amount, ...(method ? { method } : {}) }); | ||
| } | ||
| return out; | ||
| } | ||
|
|
||
| async function readJson(path: string): Promise<unknown | null> { | ||
| try { | ||
| return JSON.parse(await readFile(path, "utf8")); | ||
| } catch { | ||
| return null; | ||
| } | ||
| } | ||
|
|
||
| const candidates: ChargeCandidate[] = []; | ||
| const plaidDoc = await readJson(plaidPath); | ||
| if (plaidDoc) candidates.push(...plaidCandidates(plaidDoc)); | ||
| else console.error("gather-charges: no readable Plaid transactions at " + plaidPath); | ||
|
|
||
| const gmailDoc = await readJson(gmailPath); | ||
| if (gmailDoc) { | ||
| const parsed = parseChargeCandidatesExport(JSON.stringify(gmailDoc)); | ||
| candidates.push(...(parsed ?? []).filter((c) => c.source === "gmail")); | ||
|
Comment on lines
+90
to
+91
There was a problem hiding this comment.
When Useful? React with 👍 / 👎. |
||
| } else { | ||
| console.error("gather-charges: no readable Gmail alerts at " + gmailPath); | ||
| } | ||
|
|
||
| await mkdir(dirname(outPath), { recursive: true }); | ||
| const tmp = outPath + ".tmp"; | ||
| await writeFile(tmp, JSON.stringify({ generatedAt: new Date().toISOString(), candidates }, null, 2)); | ||
| await rename(tmp, outPath); | ||
|
Comment on lines
+98
to
+99
There was a problem hiding this comment.
When both input files are missing, unreadable, or invalid JSON, the producer still exits successfully and overwrites the last-known-good export with Useful? React with 👍 / 👎. |
||
| console.log("gather-charges: wrote " + candidates.length + " candidates to " + outPath); | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| // charges-collect (#1057): the collector-side half of the charges | ||
| // auto-matcher. The pure matching core lives in src/charges-match.ts; | ||
| // this module owns everything stateful or parsing-shaped around it: | ||
| // | ||
| // - the charges export contract: a Muse-side producer (an agent | ||
| // with the Plaid and Gmail connectors, which the collector process | ||
| // cannot reach in-process) writes ChargeCandidate[] JSON to | ||
| // ~/atlas/charges-export/latest.json — the file-export convention | ||
| // gather-quota.ts family 2 established (#1301). The collector | ||
| // ingests that file; an absent or malformed export changes | ||
| // nothing (the prose extractor stays in charge). | ||
| // - building the matcher's service refs from the hub's own SaaS | ||
| // service list, so a charge matches the services the hub tracks. | ||
| // - the persisted "already surfaced" state for unmatched SaaS | ||
| // charges: an unmatched charge joins the feed's chargesUnmatched | ||
| // section exactly once; the collector records its key in a state | ||
| // file and never surfaces the same charge again. Re-running the | ||
| // collector for the same period is a no-op for surfacing. | ||
| // | ||
| // No balances or debt data ever enter here: candidates are charge | ||
| // evidence only (date, merchant, amount, payment-method label). | ||
|
|
||
| import type { ChargeCandidate, ChargeServiceRef, UnmatchedCharge } from "./charges-match.ts"; | ||
| import type { SaasService } from "./model.ts"; | ||
|
|
||
| /** | ||
| * Parse a charges export document into candidates. Returns null when | ||
| * the document is not JSON at all (the caller keeps its fallback); | ||
| * a JSON document with malformed entries yields the valid subset — | ||
| * one bad row from a producer must not sink the whole period. | ||
| */ | ||
| export function parseChargeCandidatesExport(text: string): ChargeCandidate[] | null { | ||
| let parsed: unknown; | ||
| try { | ||
| parsed = JSON.parse(text); | ||
| } catch { | ||
| return null; | ||
| } | ||
| const list = Array.isArray(parsed) | ||
| ? parsed | ||
| : (parsed as { candidates?: unknown })?.candidates; | ||
| if (!Array.isArray(list)) return null; | ||
| const out: ChargeCandidate[] = []; | ||
| for (const entry of list) { | ||
| if (typeof entry !== "object" || entry === null) continue; | ||
| const e = entry as Record<string, unknown>; | ||
| if (e.source !== "plaid" && e.source !== "gmail") continue; | ||
| if (typeof e.at !== "string" || !Number.isFinite(Date.parse(e.at))) continue; | ||
| if (typeof e.merchant !== "string" || !e.merchant.trim()) continue; | ||
| if (typeof e.amountUsd !== "number" || !Number.isFinite(e.amountUsd)) continue; | ||
| out.push({ | ||
| source: e.source, | ||
| at: e.at, | ||
| merchant: e.merchant, | ||
| amountUsd: e.amountUsd, | ||
| ...(typeof e.method === "string" && e.method.trim() ? { method: e.method } : {}), | ||
| }); | ||
| } | ||
| return out; | ||
| } | ||
|
|
||
| /** The matcher's service refs: the hub's tracked SaaS services by name. */ | ||
| export function serviceRefsFromSaas(saas: SaasService[]): ChargeServiceRef[] { | ||
| return saas.map((s) => ({ name: s.name })); | ||
| } | ||
|
|
||
| /** | ||
| * Identity of one unmatched charge for surfacing purposes: the same | ||
| * charge re-observed next period (a new posting date) is a new | ||
| * charge and may surface again; the same posting re-collected is not. | ||
| */ | ||
| export function unmatchedChargeKey(u: UnmatchedCharge): string { | ||
| return `${u.normalized}|${u.amountUsd.toFixed(2)}|${u.at}`; | ||
|
Comment on lines
+72
to
+73
There was a problem hiding this comment.
When a Gmail alert is collected before its corresponding Plaid posting, the same purchase surfaces twice if their dates differ. For example, a Gmail Cursor charge dated October 26 is recorded under that exact date; when the matching Plaid charge dated October 28 arrives, Useful? React with 👍 / 👎. |
||
| } | ||
|
|
||
| export interface SurfacingPlan { | ||
| /** Unmatched charges not present in the persisted state. */ | ||
| fresh: UnmatchedCharge[]; | ||
| /** The full key set to persist (prior keys plus the fresh ones). */ | ||
| nextKeys: string[]; | ||
| } | ||
|
|
||
| /** | ||
| * Decide which unmatched charges surface this run. Pure: the | ||
| * collector loads the persisted keys, plans, publishes `fresh`, and | ||
| * writes `nextKeys` back — in that order, so a crash between publish | ||
| * and persist can re-surface (visible, harmless) but never lose a | ||
| * surfacing silently. | ||
| */ | ||
| export function planUnmatchedSurfacing(unmatched: UnmatchedCharge[], surfacedKeys: string[]): SurfacingPlan { | ||
| const seen = new Set(surfacedKeys); | ||
| const fresh: UnmatchedCharge[] = []; | ||
| for (const u of unmatched) { | ||
| const key = unmatchedChargeKey(u); | ||
| if (seen.has(key)) continue; | ||
| seen.add(key); | ||
| fresh.push(u); | ||
| } | ||
| return { fresh, nextKeys: [...seen] }; | ||
| } | ||
|
|
||
| /** Parse the persisted surfaced-keys state file; anything odd = empty state. */ | ||
| export function parseSurfacedState(text: string): string[] { | ||
| try { | ||
| const parsed: unknown = JSON.parse(text); | ||
| const keys = (parsed as { surfacedKeys?: unknown })?.surfacedKeys; | ||
| return Array.isArray(keys) ? keys.filter((k): k is string => typeof k === "string") : []; | ||
| } catch { | ||
| return []; | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If the laptop is unreachable during a collection, these keys are permanently marked as surfaced before the feed reaches its consumer. In
scripts/push-fleet-feed.sh, the collector runs before the Python transfer section; if that transfer fails, the next timer run recollects the feed, excludes the already-recorded charges, and overwrites the only feed containing their alerts. Consequently, those unmatched charges never appear on the laptop. Retain pending alerts across collections or acknowledge their keys only after successful delivery.Useful? React with 👍 / 👎.