From fbf1e52d30f6b486d0f159d52927b8432c3942c9 Mon Sep 17 00:00:00 2001 From: AOJDevStudio Date: Fri, 5 Jun 2026 11:40:47 -0500 Subject: [PATCH 1/3] feat(aoj-459): add event trigger path --- apps/simplefin-sync/package.json | 5 +- apps/simplefin-sync/src/trigger.test.ts | 176 ++++++++++++ apps/simplefin-sync/src/trigger.ts | 297 ++++++++++++++++++++ buy_ticket_agent/pipeline.py | 28 +- tests/python/test_buy_ticket_agent_smoke.py | 34 +++ 5 files changed, 538 insertions(+), 2 deletions(-) create mode 100644 apps/simplefin-sync/src/trigger.test.ts create mode 100644 apps/simplefin-sync/src/trigger.ts diff --git a/apps/simplefin-sync/package.json b/apps/simplefin-sync/package.json index e3b5185d..f7e2499e 100644 --- a/apps/simplefin-sync/package.json +++ b/apps/simplefin-sync/package.json @@ -5,7 +5,10 @@ "type": "module", "scripts": { "claim": "bun run src/claim.ts", - "accounts": "bun run src/accounts.ts" + "accounts": "bun run src/accounts.ts", + "trigger": "bun run src/trigger.ts", + "test": "bun test", + "typecheck": "tsc --noEmit" }, "devDependencies": { "@types/bun": "latest", diff --git a/apps/simplefin-sync/src/trigger.test.ts b/apps/simplefin-sync/src/trigger.test.ts new file mode 100644 index 00000000..be52497b --- /dev/null +++ b/apps/simplefin-sync/src/trigger.test.ts @@ -0,0 +1,176 @@ +import { mkdtemp, readdir, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; + +import { describe, expect, test } from "bun:test"; + +import type { SimpleFinClient } from "./client"; +import type { SfinAccount, SfinAccountSet, SfinTransaction } from "./types"; +import { + buildBuyTicketCommand, + buildBuyTicketEnv, + createMemorySeenStore, + findNewDepositDetections, + pollDepositTriggers, +} from "./trigger"; + +function transaction(overrides: Partial): SfinTransaction { + return { + id: "tx-default", + posted: 1_779_984_000, + amount: "0.00", + description: "test transaction", + ...overrides, + }; +} + +function account(transactions: SfinTransaction[]): SfinAccount { + return { + id: "account-default", + name: "Test source", + currency: "USD", + balance: "100.00", + "balance-date": 1_779_984_000, + org: { name: "Test org" }, + transactions, + }; +} + +function accountSet(transactions: SfinTransaction[]): SfinAccountSet { + return { + errors: [], + accounts: [account(transactions)], + }; +} + +describe("SimpleFIN deposit trigger", () => { + test("detects only deposits over the threshold with sanitized keys", async () => { + const seenStore = createMemorySeenStore(); + const { detections, skippedDuplicates } = await findNewDepositDetections( + accountSet([ + transaction({ id: "tx-3100", amount: "3100.00" }), + transaction({ id: "tx-threshold", amount: "3000.00" }), + transaction({ id: "tx-debit", amount: "-5000.00" }), + ]), + seenStore, + ); + + expect(skippedDuplicates).toBe(0); + expect(detections).toHaveLength(1); + expect(detections[0].amount).toBe("3100.00"); + expect(detections[0].transactionKey).not.toContain("tx-3100"); + expect(detections[0].sourceAccountKey).not.toContain("account-default"); + }); + + test("suppresses duplicate transaction ids within a single poll response", async () => { + const seenStore = createMemorySeenStore(); + const { detections, skippedDuplicates } = await findNewDepositDetections( + accountSet([ + transaction({ id: "tx-duplicate", amount: "3100.00" }), + transaction({ id: "tx-duplicate", amount: "3100.00" }), + ]), + seenStore, + ); + + expect(detections).toHaveLength(1); + expect(skippedDuplicates).toBe(1); + }); + + test("polling triggers once and suppresses duplicate transaction ids", async () => { + const projectRoot = await mkdtemp(path.join(tmpdir(), "simplefin-trigger-")); + const seenStore = createMemorySeenStore(); + const calls: string[] = []; + const client: SimpleFinClient = { + baseUrl: "https://simplefin.example.test", + async fetchAccounts() { + return accountSet([transaction({ id: "tx-1", amount: "3100.00" })]); + }, + }; + + try { + const first = await pollDepositTriggers({ + client, + seenStore, + projectRoot, + now: new Date("2026-06-05T12:00:00.000Z"), + trigger: async (detection) => { + calls.push(detection.transactionKey); + return { + status: "triggered", + exitCode: 0, + stdoutChars: 10, + stderrChars: 0, + }; + }, + }); + const second = await pollDepositTriggers({ + client, + seenStore, + projectRoot, + now: new Date("2026-06-05T16:00:00.000Z"), + trigger: async (detection) => { + calls.push(detection.transactionKey); + return { + status: "triggered", + exitCode: 0, + stdoutChars: 10, + stderrChars: 0, + }; + }, + }); + + expect(first).toMatchObject({ + detections: 1, + triggered: 1, + skippedDuplicates: 0, + failed: 0, + }); + expect(second).toMatchObject({ + detections: 0, + triggered: 0, + skippedDuplicates: 1, + failed: 0, + }); + expect(calls).toHaveLength(1); + + const logDir = path.join(projectRoot, "notebooks", "auto-tickets", "runs"); + const [logFile] = await readdir(logDir); + const log = JSON.parse(await readFile(path.join(logDir, logFile), "utf8")); + expect(log).toMatchObject({ + event: "simplefin_deposit_detection", + status: "triggered", + amount: "3100.00", + }); + expect(log.transaction_key).toBe(calls[0]); + expect(log.transaction_key).not.toContain("tx-1"); + } finally { + await rm(projectRoot, { recursive: true, force: true }); + } + }); + + test("builds the buy-ticket subprocess command and sanitized trigger env", () => { + const detection = { + amount: "3100.00", + transactionKey: "transaction-key", + sourceAccountKey: "source-account-key", + posted: 1_779_984_000, + pending: false, + }; + + expect(buildBuyTicketCommand()).toEqual([ + "uv", + "run", + "python", + "-m", + "buy_ticket_agent.main", + "--smoke", + ]); + expect(buildBuyTicketEnv(detection, { PATH: "/bin" })).toMatchObject({ + PATH: "/bin", + BUY_TICKET_TRIGGER_SOURCE: "simplefin_deposit", + BUY_TICKET_TRIGGER_AMOUNT: "3100.00", + BUY_TICKET_TRIGGER_ACCOUNT_KEY: "source-account-key", + BUY_TICKET_TRIGGER_TRANSACTION_KEY: "transaction-key", + }); + }); +}); diff --git a/apps/simplefin-sync/src/trigger.ts b/apps/simplefin-sync/src/trigger.ts new file mode 100644 index 00000000..299ec698 --- /dev/null +++ b/apps/simplefin-sync/src/trigger.ts @@ -0,0 +1,297 @@ +#!/usr/bin/env bun +/** + * SimpleFIN deposit trigger for the buy-ticket smoke pipeline. + */ +import { createHash } from "node:crypto"; +import { mkdir, writeFile } from "node:fs/promises"; +import path from "node:path"; + +import { createClient, toSfinTimestamp, type SimpleFinClient } from "./client"; +import type { SfinAccount, SfinAccountSet, SfinTransaction } from "./types"; + +const DEFAULT_THRESHOLD = "3000.00"; +const MONEY_SCALE = 4; +const DEFAULT_POLL_INTERVAL_MS = 4 * 60 * 60 * 1000; + +export interface DepositDetection { + transactionKey: string; + sourceAccountKey: string; + amount: string; + posted: number; + pending: boolean; +} + +export interface TriggerResult { + status: "triggered" | "failed"; + exitCode: number; + stdoutChars: number; + stderrChars: number; +} + +export interface SeenTransactionStore { + has(key: string): boolean | Promise; + mark(key: string): void | Promise; +} + +export interface PollOptions { + client: SimpleFinClient; + seenStore: SeenTransactionStore; + projectRoot: string; + threshold?: string; + trigger?: (detection: DepositDetection) => Promise; + now?: Date; + lookbackDays?: number; +} + +export interface PollResult { + scannedAccounts: number; + detections: number; + triggered: number; + skippedDuplicates: number; + failed: number; +} + +export function createMemorySeenStore( + initialKeys: Iterable = [], +): SeenTransactionStore { + const seen = new Set(initialKeys); + return { + has(key: string) { + return seen.has(key); + }, + mark(key: string) { + seen.add(key); + }, + }; +} + +function parseDecimalUnits(value: string): bigint | null { + const match = value.trim().match(/^([+-]?)(\d+)(?:\.(\d+))?$/); + if (!match) return null; + + const [, sign, whole, fractionRaw = ""] = match; + const extra = fractionRaw.slice(MONEY_SCALE); + if (/[1-9]/.test(extra)) return null; + + const fraction = fractionRaw.slice(0, MONEY_SCALE).padEnd(MONEY_SCALE, "0"); + const units = BigInt(whole) * 10_000n + BigInt(fraction || "0"); + return sign === "-" ? -units : units; +} + +export function isDepositOverThreshold( + transaction: SfinTransaction, + threshold = DEFAULT_THRESHOLD, +): boolean { + const amount = parseDecimalUnits(transaction.amount); + const thresholdUnits = parseDecimalUnits(threshold); + if (amount === null || thresholdUnits === null) return false; + return amount > thresholdUnits; +} + +function stableKey(...parts: string[]): string { + return createHash("sha256").update(parts.join("\u001f")).digest("hex").slice(0, 16); +} + +export function detectionFromTransaction( + account: SfinAccount, + transaction: SfinTransaction, + threshold = DEFAULT_THRESHOLD, +): DepositDetection | null { + if (!isDepositOverThreshold(transaction, threshold)) return null; + return { + transactionKey: stableKey(account.id, transaction.id), + sourceAccountKey: stableKey(account.id), + amount: transaction.amount, + posted: transaction.posted, + pending: transaction.pending ?? false, + }; +} + +export async function findNewDepositDetections( + accountSet: SfinAccountSet, + seenStore: SeenTransactionStore, + threshold = DEFAULT_THRESHOLD, +): Promise<{ detections: DepositDetection[]; skippedDuplicates: number }> { + const detections: DepositDetection[] = []; + const queuedThisPoll = new Set(); + let skippedDuplicates = 0; + + for (const account of accountSet.accounts) { + for (const transaction of account.transactions ?? []) { + const detection = detectionFromTransaction(account, transaction, threshold); + if (!detection) continue; + + if ( + queuedThisPoll.has(detection.transactionKey) || + (await seenStore.has(detection.transactionKey)) + ) { + skippedDuplicates += 1; + continue; + } + queuedThisPoll.add(detection.transactionKey); + detections.push(detection); + } + } + + return { detections, skippedDuplicates }; +} + +export function buildBuyTicketCommand(): string[] { + return ["uv", "run", "python", "-m", "buy_ticket_agent.main", "--smoke"]; +} + +export function buildBuyTicketEnv( + detection: DepositDetection, + baseEnv: NodeJS.ProcessEnv = process.env, +): NodeJS.ProcessEnv { + return { + ...baseEnv, + BUY_TICKET_TRIGGER_SOURCE: "simplefin_deposit", + BUY_TICKET_TRIGGER_AMOUNT: detection.amount, + BUY_TICKET_TRIGGER_ACCOUNT_KEY: detection.sourceAccountKey, + BUY_TICKET_TRIGGER_TRANSACTION_KEY: detection.transactionKey, + }; +} + +async function countStreamChars(stream: ReadableStream): Promise { + const decoder = new TextDecoder(); + const reader = stream.getReader(); + let chars = 0; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + chars += decoder.decode(value, { stream: true }).length; + } + + chars += decoder.decode().length; + return chars; +} + +export async function runBuyTicketSubprocess( + detection: DepositDetection, + projectRoot: string, +): Promise { + const proc = Bun.spawn(buildBuyTicketCommand(), { + cwd: projectRoot, + env: buildBuyTicketEnv(detection), + stdout: "pipe", + stderr: "pipe", + }); + + const [stdoutChars, stderrChars, exitCode] = await Promise.all([ + countStreamChars(proc.stdout), + countStreamChars(proc.stderr), + proc.exited, + ]); + + return { + status: exitCode === 0 ? "triggered" : "failed", + exitCode, + stdoutChars, + stderrChars, + }; +} + +async function writeTriggerLog( + projectRoot: string, + detection: DepositDetection, + result: TriggerResult, + now: Date, +): Promise { + const logDir = path.join(projectRoot, "notebooks", "auto-tickets", "runs"); + const logPath = path.join( + logDir, + `simplefin-trigger-${now.toISOString().replace(/[:.]/g, "-")}-${detection.transactionKey}.json`, + ); + const payload = { + event: "simplefin_deposit_detection", + created_at: now.toISOString(), + status: result.status, + amount: detection.amount, + source_account_key: detection.sourceAccountKey, + transaction_key: detection.transactionKey, + posted: detection.posted, + pending: detection.pending, + buy_ticket: result, + }; + + await mkdir(logDir, { recursive: true }); + await writeFile(logPath, `${JSON.stringify(payload, null, 2)}\n`); +} + +export async function pollDepositTriggers(options: PollOptions): Promise { + const now = options.now ?? new Date(); + const lookbackDays = options.lookbackDays ?? 7; + const startDate = toSfinTimestamp( + new Date(now.valueOf() - lookbackDays * 24 * 60 * 60 * 1000), + ); + const accountSet = await options.client.fetchAccounts({ + startDate, + pending: true, + }); + const { detections, skippedDuplicates } = await findNewDepositDetections( + accountSet, + options.seenStore, + options.threshold, + ); + const trigger = + options.trigger ?? + ((detection: DepositDetection) => + runBuyTicketSubprocess(detection, options.projectRoot)); + + let triggered = 0; + let failed = 0; + for (const detection of detections) { + const result = await trigger(detection); + await writeTriggerLog(options.projectRoot, detection, result, now); + if (result.status === "triggered") { + await options.seenStore.mark(detection.transactionKey); + triggered += 1; + } else { + failed += 1; + } + } + + return { + scannedAccounts: accountSet.accounts.length, + detections: detections.length, + triggered, + skippedDuplicates, + failed, + }; +} + +async function runCli(argv: string[]): Promise { + const accessUrl = process.env.SIMPLEFIN_ACCESS_URL?.trim(); + if (!accessUrl) { + console.error("SIMPLEFIN_ACCESS_URL is empty. Run `bun run claim` first."); + return 1; + } + + const projectRoot = path.resolve(import.meta.dir, "..", "..", ".."); + const client = createClient(accessUrl); + const seenStore = createMemorySeenStore(); + const watch = argv.includes("--watch"); + const intervalMs = + Number(process.env.SIMPLEFIN_TRIGGER_INTERVAL_MS ?? "") || + DEFAULT_POLL_INTERVAL_MS; + + do { + const result = await pollDepositTriggers({ + client, + seenStore, + projectRoot, + }); + console.log(JSON.stringify(result)); + if (!watch) break; + await Bun.sleep(intervalMs); + } while (true); + + return 0; +} + +if (import.meta.main) { + const exitCode = await runCli(Bun.argv.slice(2)); + process.exit(exitCode); +} diff --git a/buy_ticket_agent/pipeline.py b/buy_ticket_agent/pipeline.py index 5888d598..3dd39384 100644 --- a/buy_ticket_agent/pipeline.py +++ b/buy_ticket_agent/pipeline.py @@ -70,6 +70,24 @@ def _output_length(output: str | bytes | None) -> int: return len(output) +def _trigger_context_from_env() -> dict[str, str] | None: + source = os.getenv("BUY_TICKET_TRIGGER_SOURCE") + if not source: + return None + + context = {"source": source} + optional_fields = { + "BUY_TICKET_TRIGGER_AMOUNT": "amount", + "BUY_TICKET_TRIGGER_ACCOUNT_KEY": "source_account_key", + "BUY_TICKET_TRIGGER_TRANSACTION_KEY": "transaction_key", + } + for env_name, field_name in optional_fields.items(): + value = os.getenv(env_name) + if value: + context[field_name] = value + return context + + def run_layer3_smoke_cli(project_root: Path) -> CliEnvelope: """Invoke the existing ITC Risk CLI once for the smoke path.""" command_args = ( @@ -133,8 +151,9 @@ def _make_ticket_payload( ticket_id: str, created_at: str, cli_result: CliEnvelope, + trigger_context: dict[str, str] | None, ) -> dict: - return { + payload = { "id": ticket_id, "run_id": run_id, "created_at": created_at, @@ -151,6 +170,9 @@ def _make_ticket_payload( }, "disclaimer": DISCLAIMER, } + if trigger_context is not None: + payload["trigger"] = trigger_context + return payload def run_smoke(project_root: Path | None = None) -> SmokeResult: @@ -162,6 +184,7 @@ def run_smoke(project_root: Path | None = None) -> SmokeResult: ticket_id = f"ticket-{run_id}" notification_config = resolve_notification_config() + trigger_context = _trigger_context_from_env() initialize_state(paths.state_db) cli_result = run_layer3_smoke_cli(paths.project_root) @@ -179,6 +202,7 @@ def run_smoke(project_root: Path | None = None) -> SmokeResult: "status": status, "draft_path": None, "state_db": _relative(paths.state_db, paths.project_root), + "trigger": trigger_context, "layer3": asdict(cli_result), "notification": None, } @@ -205,6 +229,7 @@ def run_smoke(project_root: Path | None = None) -> SmokeResult: ticket_id=ticket_id, created_at=created_at, cli_result=cli_result, + trigger_context=trigger_context, ) _write_json(draft_path, ticket_payload) @@ -227,6 +252,7 @@ def run_smoke(project_root: Path | None = None) -> SmokeResult: "status": status, "draft_path": _relative(draft_path, paths.project_root), "state_db": _relative(paths.state_db, paths.project_root), + "trigger": trigger_context, "layer3": asdict(cli_result), "notification": asdict(notification), } diff --git a/tests/python/test_buy_ticket_agent_smoke.py b/tests/python/test_buy_ticket_agent_smoke.py index 9b32dbda..7d79bedf 100644 --- a/tests/python/test_buy_ticket_agent_smoke.py +++ b/tests/python/test_buy_ticket_agent_smoke.py @@ -251,6 +251,40 @@ def test_run_smoke_writes_draft_log_and_state(tmp_path: Path, monkeypatch) -> No assert (run_count, ticket_count, decision_count) == (1, 1, 1) +def test_run_smoke_includes_sanitized_trigger_context( + tmp_path: Path, monkeypatch +) -> None: + """Smoke artifacts include trigger context passed by the event poller.""" + monkeypatch.delenv("BUY_TICKET_AGENT_NTFY_TOPIC_SECRET_ID", raising=False) + monkeypatch.delenv("NTFY_TOPIC_SECRET_ID", raising=False) + monkeypatch.delenv("BUY_TICKET_AGENT_NTFY_TOPIC", raising=False) + monkeypatch.delenv("NTFY_TOPIC", raising=False) + monkeypatch.setenv("BUY_TICKET_TRIGGER_SOURCE", "simplefin_deposit") + monkeypatch.setenv("BUY_TICKET_TRIGGER_AMOUNT", "3100.00") + monkeypatch.setenv("BUY_TICKET_TRIGGER_ACCOUNT_KEY", "account-key") + monkeypatch.setenv("BUY_TICKET_TRIGGER_TRANSACTION_KEY", "transaction-key") + + completed = MagicMock() + completed.returncode = 0 + completed.stdout = '{"symbol":"SP500","current_risk_score":0.4}' + completed.stderr = "" + + with patch("buy_ticket_agent.pipeline.subprocess.run", return_value=completed): + result = run_smoke(project_root=tmp_path) + + draft = json.loads((tmp_path / result.draft_path).read_text()) + log = json.loads((tmp_path / result.log_path).read_text()) + expected_context = { + "source": "simplefin_deposit", + "amount": "3100.00", + "source_account_key": "account-key", + "transaction_key": "transaction-key", + } + + assert draft["trigger"] == expected_context + assert log["trigger"] == expected_context + + def test_run_smoke_does_not_record_success_state_if_log_write_fails( tmp_path: Path, monkeypatch ) -> None: From b29cd5c58d37ed8b2ee1301c8c0cb2e1d72aa51f Mon Sep 17 00:00:00 2001 From: AOJDevStudio Date: Fri, 5 Jun 2026 12:00:14 -0500 Subject: [PATCH 2/3] fix(aoj-459): harden event trigger path --- apps/simplefin-sync/src/trigger.test.ts | 126 +++++++++++++- apps/simplefin-sync/src/trigger.ts | 221 +++++++++++++++++++++--- 2 files changed, 321 insertions(+), 26 deletions(-) diff --git a/apps/simplefin-sync/src/trigger.test.ts b/apps/simplefin-sync/src/trigger.test.ts index be52497b..72dc7145 100644 --- a/apps/simplefin-sync/src/trigger.test.ts +++ b/apps/simplefin-sync/src/trigger.test.ts @@ -1,4 +1,4 @@ -import { mkdtemp, readdir, readFile, rm } from "node:fs/promises"; +import { mkdtemp, readdir, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -9,6 +9,7 @@ import type { SfinAccount, SfinAccountSet, SfinTransaction } from "./types"; import { buildBuyTicketCommand, buildBuyTicketEnv, + createFileSeenStore, createMemorySeenStore, findNewDepositDetections, pollDepositTriggers, @@ -135,7 +136,7 @@ describe("SimpleFIN deposit trigger", () => { const logDir = path.join(projectRoot, "notebooks", "auto-tickets", "runs"); const [logFile] = await readdir(logDir); - const log = JSON.parse(await readFile(path.join(logDir, logFile), "utf8")); + const log = JSON.parse(await Bun.file(path.join(logDir, logFile)).text()); expect(log).toMatchObject({ event: "simplefin_deposit_detection", status: "triggered", @@ -148,6 +149,117 @@ describe("SimpleFIN deposit trigger", () => { } }); + test("persists seen transaction ids across store instances", async () => { + const projectRoot = await mkdtemp(path.join(tmpdir(), "simplefin-trigger-")); + const statePath = path.join(projectRoot, "notebooks", "auto-tickets", "seen.json"); + const client: SimpleFinClient = { + baseUrl: "https://simplefin.example.test", + async fetchAccounts() { + return accountSet([transaction({ id: "tx-1", amount: "3100.00" })]); + }, + }; + const calls: string[] = []; + + try { + await pollDepositTriggers({ + client, + seenStore: createFileSeenStore(statePath), + projectRoot, + trigger: async (detection) => { + calls.push(detection.transactionKey); + return { + status: "triggered", + exitCode: 0, + stdoutChars: 10, + stderrChars: 0, + }; + }, + }); + const second = await pollDepositTriggers({ + client, + seenStore: createFileSeenStore(statePath), + projectRoot, + trigger: async (detection) => { + calls.push(detection.transactionKey); + return { + status: "triggered", + exitCode: 0, + stdoutChars: 10, + stderrChars: 0, + }; + }, + }); + + expect(calls).toHaveLength(1); + expect(second).toMatchObject({ + detections: 0, + triggered: 0, + skippedDuplicates: 1, + }); + const state = JSON.parse(await Bun.file(statePath).text()); + expect(state.seen).toHaveLength(1); + expect(state.seen[0]).not.toContain("tx-1"); + } finally { + await rm(projectRoot, { recursive: true, force: true }); + } + }); + + test("caps repeated failed trigger attempts", async () => { + const projectRoot = await mkdtemp(path.join(tmpdir(), "simplefin-trigger-")); + const seenStore = createMemorySeenStore(); + const client: SimpleFinClient = { + baseUrl: "https://simplefin.example.test", + async fetchAccounts() { + return accountSet([transaction({ id: "tx-1", amount: "3100.00" })]); + }, + }; + const calls: string[] = []; + + try { + const trigger = async (detection: { transactionKey: string }) => { + calls.push(detection.transactionKey); + return { + status: "failed" as const, + exitCode: 1, + stdoutChars: 0, + stderrChars: 10, + }; + }; + const first = await pollDepositTriggers({ + client, + seenStore, + projectRoot, + maxFailures: 2, + trigger, + }); + const second = await pollDepositTriggers({ + client, + seenStore, + projectRoot, + maxFailures: 2, + trigger, + }); + const third = await pollDepositTriggers({ + client, + seenStore, + projectRoot, + maxFailures: 2, + trigger, + }); + + expect(calls).toHaveLength(2); + expect(first).toMatchObject({ failed: 1, cappedFailures: 0 }); + expect(second).toMatchObject({ failed: 1, cappedFailures: 1 }); + expect(third).toMatchObject({ + detections: 0, + skippedDuplicates: 1, + failed: 0, + }); + } finally { + await rm(projectRoot, { recursive: true, force: true }); + } + }); + test("builds the buy-ticket subprocess command and sanitized trigger env", () => { const detection = { amount: "3100.00", @@ -157,6 +269,12 @@ describe("SimpleFIN deposit trigger", () => { pending: false, }; + const env = buildBuyTicketEnv(detection, { + PATH: "/bin", + SIMPLEFIN_ACCESS_URL: "credential-bearing-url", + SIMPLEFIN_TRIGGER_INTERVAL_MS: "1000", + }); + expect(buildBuyTicketCommand()).toEqual([ "uv", "run", @@ -165,12 +283,14 @@ describe("SimpleFIN deposit trigger", () => { "buy_ticket_agent.main", "--smoke", ]); - expect(buildBuyTicketEnv(detection, { PATH: "/bin" })).toMatchObject({ + expect(env).toMatchObject({ PATH: "/bin", BUY_TICKET_TRIGGER_SOURCE: "simplefin_deposit", BUY_TICKET_TRIGGER_AMOUNT: "3100.00", BUY_TICKET_TRIGGER_ACCOUNT_KEY: "source-account-key", BUY_TICKET_TRIGGER_TRANSACTION_KEY: "transaction-key", }); + expect(env).not.toHaveProperty("SIMPLEFIN_ACCESS_URL"); + expect(env).not.toHaveProperty("SIMPLEFIN_TRIGGER_INTERVAL_MS"); }); }); diff --git a/apps/simplefin-sync/src/trigger.ts b/apps/simplefin-sync/src/trigger.ts index 299ec698..214ce09e 100644 --- a/apps/simplefin-sync/src/trigger.ts +++ b/apps/simplefin-sync/src/trigger.ts @@ -3,7 +3,7 @@ * SimpleFIN deposit trigger for the buy-ticket smoke pipeline. */ import { createHash } from "node:crypto"; -import { mkdir, writeFile } from "node:fs/promises"; +import { mkdir } from "node:fs/promises"; import path from "node:path"; import { createClient, toSfinTimestamp, type SimpleFinClient } from "./client"; @@ -12,6 +12,9 @@ import type { SfinAccount, SfinAccountSet, SfinTransaction } from "./types"; const DEFAULT_THRESHOLD = "3000.00"; const MONEY_SCALE = 4; const DEFAULT_POLL_INTERVAL_MS = 4 * 60 * 60 * 1000; +const DEFAULT_TRIGGER_TIMEOUT_MS = 10 * 60 * 1000; +const DEFAULT_MAX_TRIGGER_FAILURES = 3; +const CHILD_ENV_DENYLIST_PREFIXES = ["SIMPLEFIN_"]; export interface DepositDetection { transactionKey: string; @@ -26,11 +29,14 @@ export interface TriggerResult { exitCode: number; stdoutChars: number; stderrChars: number; + timedOut?: boolean; } export interface SeenTransactionStore { has(key: string): boolean | Promise; mark(key: string): void | Promise; + failureCount?(key: string): number | Promise; + markFailure?(key: string): number | Promise; } export interface PollOptions { @@ -41,6 +47,7 @@ export interface PollOptions { trigger?: (detection: DepositDetection) => Promise; now?: Date; lookbackDays?: number; + maxFailures?: number; } export interface PollResult { @@ -49,18 +56,96 @@ export interface PollResult { triggered: number; skippedDuplicates: number; failed: number; + cappedFailures: number; +} + +interface SeenStateFile { + seen?: string[]; + failures?: Record; } export function createMemorySeenStore( initialKeys: Iterable = [], ): SeenTransactionStore { const seen = new Set(initialKeys); + const failures = new Map(); return { has(key: string) { return seen.has(key); }, mark(key: string) { seen.add(key); + failures.delete(key); + }, + failureCount(key: string) { + return failures.get(key) ?? 0; + }, + markFailure(key: string) { + const count = (failures.get(key) ?? 0) + 1; + failures.set(key, count); + return count; + }, + }; +} + +export function createFileSeenStore(statePath: string): SeenTransactionStore { + const seen = new Set(); + const failures = new Map(); + let loaded = false; + + async function load(): Promise { + if (loaded) return; + const file = Bun.file(statePath); + if (await file.exists()) { + const parsed = JSON.parse(await file.text()) as SeenStateFile; + for (const key of parsed.seen ?? []) { + seen.add(key); + } + for (const [key, count] of Object.entries(parsed.failures ?? {})) { + if (Number.isInteger(count) && count > 0) { + failures.set(key, count); + } + } + } + loaded = true; + } + + async function persist(): Promise { + await mkdir(path.dirname(statePath), { recursive: true }); + await Bun.write( + statePath, + `${JSON.stringify( + { + seen: [...seen].sort(), + failures: Object.fromEntries([...failures].sort()), + }, + null, + 2, + )}\n`, + ); + } + + return { + async has(key: string) { + await load(); + return seen.has(key); + }, + async mark(key: string) { + await load(); + seen.add(key); + failures.delete(key); + await persist(); + }, + async failureCount(key: string) { + await load(); + return failures.get(key) ?? 0; + }, + async markFailure(key: string) { + await load(); + const count = (failures.get(key) ?? 0) + 1; + failures.set(key, count); + await persist(); + return count; }, }; } @@ -140,12 +225,23 @@ export function buildBuyTicketCommand(): string[] { return ["uv", "run", "python", "-m", "buy_ticket_agent.main", "--smoke"]; } +function childProcessEnv(baseEnv: NodeJS.ProcessEnv): NodeJS.ProcessEnv { + const env: NodeJS.ProcessEnv = {}; + for (const [key, value] of Object.entries(baseEnv)) { + if (CHILD_ENV_DENYLIST_PREFIXES.some((prefix) => key.startsWith(prefix))) { + continue; + } + env[key] = value; + } + return env; +} + export function buildBuyTicketEnv( detection: DepositDetection, baseEnv: NodeJS.ProcessEnv = process.env, ): NodeJS.ProcessEnv { return { - ...baseEnv, + ...childProcessEnv(baseEnv), BUY_TICKET_TRIGGER_SOURCE: "simplefin_deposit", BUY_TICKET_TRIGGER_AMOUNT: detection.amount, BUY_TICKET_TRIGGER_ACCOUNT_KEY: detection.sourceAccountKey, @@ -171,6 +267,7 @@ async function countStreamChars(stream: ReadableStream): Promise { const proc = Bun.spawn(buildBuyTicketCommand(), { cwd: projectRoot, @@ -178,19 +275,29 @@ export async function runBuyTicketSubprocess( stdout: "pipe", stderr: "pipe", }); - - const [stdoutChars, stderrChars, exitCode] = await Promise.all([ - countStreamChars(proc.stdout), - countStreamChars(proc.stderr), - proc.exited, - ]); - - return { - status: exitCode === 0 ? "triggered" : "failed", - exitCode, - stdoutChars, - stderrChars, - }; + let timedOut = false; + const timeout = setTimeout(() => { + timedOut = true; + proc.kill(); + }, timeoutMs); + + try { + const [stdoutChars, stderrChars, exitCode] = await Promise.all([ + countStreamChars(proc.stdout), + countStreamChars(proc.stderr), + proc.exited, + ]); + + return { + status: exitCode === 0 && !timedOut ? "triggered" : "failed", + exitCode, + stdoutChars, + stderrChars, + ...(timedOut ? { timedOut } : {}), + }; + } finally { + clearTimeout(timeout); + } } async function writeTriggerLog( @@ -217,12 +324,27 @@ async function writeTriggerLog( }; await mkdir(logDir, { recursive: true }); - await writeFile(logPath, `${JSON.stringify(payload, null, 2)}\n`); + await Bun.write(logPath, `${JSON.stringify(payload, null, 2)}\n`); +} + +async function failureCount( + seenStore: SeenTransactionStore, + key: string, +): Promise { + return seenStore.failureCount ? await seenStore.failureCount(key) : 0; +} + +async function markFailure( + seenStore: SeenTransactionStore, + key: string, +): Promise { + return seenStore.markFailure ? await seenStore.markFailure(key) : 0; } export async function pollDepositTriggers(options: PollOptions): Promise { const now = options.now ?? new Date(); const lookbackDays = options.lookbackDays ?? 7; + const maxFailures = options.maxFailures ?? DEFAULT_MAX_TRIGGER_FAILURES; const startDate = toSfinTimestamp( new Date(now.valueOf() - lookbackDays * 24 * 60 * 60 * 1000), ); @@ -242,13 +364,28 @@ export async function pollDepositTriggers(options: PollOptions): Promise 0 && + (await failureCount(options.seenStore, detection.transactionKey)) >= maxFailures + ) { + await options.seenStore.mark(detection.transactionKey); + cappedFailures += 1; + continue; + } + const result = await trigger(detection); await writeTriggerLog(options.projectRoot, detection, result, now); if (result.status === "triggered") { await options.seenStore.mark(detection.transactionKey); triggered += 1; } else { + const failures = await markFailure(options.seenStore, detection.transactionKey); + if (maxFailures > 0 && failures >= maxFailures) { + await options.seenStore.mark(detection.transactionKey); + cappedFailures += 1; + } failed += 1; } } @@ -259,6 +396,35 @@ export async function pollDepositTriggers(options: PollOptions): Promise { const projectRoot = path.resolve(import.meta.dir, "..", "..", ".."); const client = createClient(accessUrl); - const seenStore = createMemorySeenStore(); + const seenStore = createFileSeenStore( + path.join(projectRoot, "notebooks", "auto-tickets", "simplefin-seen.json"), + ); const watch = argv.includes("--watch"); const intervalMs = Number(process.env.SIMPLEFIN_TRIGGER_INTERVAL_MS ?? "") || DEFAULT_POLL_INTERVAL_MS; do { - const result = await pollDepositTriggers({ - client, - seenStore, - projectRoot, - }); - console.log(JSON.stringify(result)); + try { + const result = await pollDepositTriggers({ + client, + seenStore, + projectRoot, + }); + console.log(JSON.stringify(result)); + } catch (error) { + if (!watch) throw error; + console.error(JSON.stringify(pollErrorPayload(error))); + await Bun.sleep(intervalMs); + continue; + } if (!watch) break; await Bun.sleep(intervalMs); } while (true); From 2e42d5450313c557b59a0f56055df896984629de Mon Sep 17 00:00:00 2001 From: AOJDevStudio Date: Fri, 5 Jun 2026 12:17:42 -0500 Subject: [PATCH 3/3] fix(aoj-459): settle trigger review findings --- apps/simplefin-sync/src/trigger.test.ts | 37 +++++++++++++++++++++++++ apps/simplefin-sync/src/trigger.ts | 35 +++++++++++++++++------ 2 files changed, 64 insertions(+), 8 deletions(-) diff --git a/apps/simplefin-sync/src/trigger.test.ts b/apps/simplefin-sync/src/trigger.test.ts index 72dc7145..50e026c0 100644 --- a/apps/simplefin-sync/src/trigger.test.ts +++ b/apps/simplefin-sync/src/trigger.test.ts @@ -260,6 +260,43 @@ describe("SimpleFIN deposit trigger", () => { } }); + test("marks successful triggers even when log writing fails", async () => { + const projectRoot = await mkdtemp(path.join(tmpdir(), "simplefin-trigger-")); + const seenStore = createMemorySeenStore(); + const client: SimpleFinClient = { + baseUrl: "https://simplefin.example.test", + async fetchAccounts() { + return accountSet([transaction({ id: "tx-1", amount: "3100.00" })]); + }, + }; + let transactionKey = ""; + + try { + await Bun.write(path.join(projectRoot, "notebooks"), "not a directory"); + await expect( + pollDepositTriggers({ + client, + seenStore, + projectRoot, + trigger: async (detection) => { + transactionKey = detection.transactionKey; + return { + status: "triggered", + exitCode: 0, + stdoutChars: 10, + stderrChars: 0, + }; + }, + }), + ).rejects.toThrow(); + + expect(transactionKey).not.toBe(""); + expect(await seenStore.has(transactionKey)).toBe(true); + } finally { + await rm(projectRoot, { recursive: true, force: true }); + } + }); + test("builds the buy-ticket subprocess command and sanitized trigger env", () => { const detection = { amount: "3100.00", diff --git a/apps/simplefin-sync/src/trigger.ts b/apps/simplefin-sync/src/trigger.ts index 214ce09e..a77252fb 100644 --- a/apps/simplefin-sync/src/trigger.ts +++ b/apps/simplefin-sync/src/trigger.ts @@ -3,7 +3,7 @@ * SimpleFIN deposit trigger for the buy-ticket smoke pipeline. */ import { createHash } from "node:crypto"; -import { mkdir } from "node:fs/promises"; +import { mkdir, rename } from "node:fs/promises"; import path from "node:path"; import { createClient, toSfinTimestamp, type SimpleFinClient } from "./client"; @@ -35,8 +35,8 @@ export interface TriggerResult { export interface SeenTransactionStore { has(key: string): boolean | Promise; mark(key: string): void | Promise; - failureCount?(key: string): number | Promise; - markFailure?(key: string): number | Promise; + failureCount(key: string): number | Promise; + markFailure(key: string): number | Promise; } export interface PollOptions { @@ -112,8 +112,9 @@ export function createFileSeenStore(statePath: string): SeenTransactionStore { async function persist(): Promise { await mkdir(path.dirname(statePath), { recursive: true }); + const tempPath = `${statePath}.${process.pid}.${Date.now()}.tmp`; await Bun.write( - statePath, + tempPath, `${JSON.stringify( { seen: [...seen].sort(), @@ -123,6 +124,7 @@ export function createFileSeenStore(statePath: string): SeenTransactionStore { 2, )}\n`, ); + await rename(tempPath, statePath); } return { @@ -331,14 +333,21 @@ async function failureCount( seenStore: SeenTransactionStore, key: string, ): Promise { - return seenStore.failureCount ? await seenStore.failureCount(key) : 0; + return seenStore.failureCount(key); } async function markFailure( seenStore: SeenTransactionStore, key: string, ): Promise { - return seenStore.markFailure ? await seenStore.markFailure(key) : 0; + return seenStore.markFailure(key); +} + +function toScrubbedError(error: unknown, fallback: string): Error { + if (error instanceof Error) { + return new Error(scrubSensitiveText(error.message || fallback)); + } + return new Error(fallback); } export async function pollDepositTriggers(options: PollOptions): Promise { @@ -376,7 +385,13 @@ export async function pollDepositTriggers(options: PollOptions): Promise { }); console.log(JSON.stringify(result)); } catch (error) { - if (!watch) throw error; console.error(JSON.stringify(pollErrorPayload(error))); + if (!watch) return 1; await Bun.sleep(intervalMs); continue; }