diff --git a/apps/worker/src/alerts-render.ts b/apps/worker/src/alerts-render.ts new file mode 100644 index 0000000..df472b2 --- /dev/null +++ b/apps/worker/src/alerts-render.ts @@ -0,0 +1,53 @@ +import type { + OutboundEmail, + OutboundNotification, + PortfolioAlert, +} from "@pointup/core"; + +function icon(alert: PortfolioAlert): string { + return alert.severity === "warning" ? "⚠️" : "ℹ️"; +} + +/** Renders alerts as a compact chat notification (Slack/Discord). */ +export function renderAlertsChat( + alerts: readonly PortfolioAlert[], +): OutboundNotification { + const header = `*PointBot alerts* — ${alerts.length} item${alerts.length === 1 ? "" : "s"} need attention`; + const lines = alerts.map((a) => `${icon(a)} ${a.message}`); + const markdown = [header, ...lines].join("\n"); + const text = markdown.replace(/\*/g, ""); + return { text, markdown }; +} + +/** Renders alerts as a plain, urgent email. */ +export function renderAlertsEmail( + alerts: readonly PortfolioAlert[], + to: string, +): OutboundEmail { + const warnings = alerts.filter((a) => a.severity === "warning").length; + const subject = + warnings > 0 + ? `PointBot: ${warnings} point alert${warnings === 1 ? "" : "s"} need attention` + : `PointBot: ${alerts.length} portfolio update${alerts.length === 1 ? "" : "s"}`; + + const text = [ + "Here's what needs your attention:", + "", + ...alerts.map((a) => `- ${icon(a)} ${a.message}`), + ].join("\n"); + + const html = ` +
+

PointBot alerts

+ +
`; + + return { to, subject, text, html }; +} diff --git a/apps/worker/src/env.ts b/apps/worker/src/env.ts index b39c812..36fc9bd 100644 --- a/apps/worker/src/env.ts +++ b/apps/worker/src/env.ts @@ -50,6 +50,10 @@ const envSchema = z.object({ /** Optional chat digest delivery — Slack / Discord incoming webhooks. */ SLACK_WEBHOOK_URL: z.url().optional(), DISCORD_WEBHOOK_URL: z.url().optional(), + + /** Proactive-alerts thresholds (the `alerts` job); core defaults apply. */ + ALERT_EXPIRY_WARNING_DAYS: z.coerce.number().int().positive().optional(), + ALERT_BIG_CHANGE_PERCENT: z.coerce.number().positive().optional(), }); export type WorkerEnv = z.infer; diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 5108b8d..39f5fe3 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -1,13 +1,14 @@ import { createContainer } from "./container"; import { loadEnv } from "./env"; import { runMigrations } from "./jobs/migrate"; +import { sendAlerts } from "./jobs/send-alerts"; import { sendDigests } from "./jobs/send-digests"; import { syncAllUsers } from "./jobs/sync-all-users"; import { createMailer } from "./mailers"; import { createNotifier } from "./notifiers"; import { createUserDirectory } from "./user-directory"; -const JOBS = ["sync", "digest", "migrate"] as const; +const JOBS = ["sync", "digest", "alerts", "migrate"] as const; type Job = (typeof JOBS)[number]; async function main(): Promise { @@ -28,6 +29,21 @@ async function main(): Promise { const container = createContainer(env); if (job === "sync") { await syncAllUsers(container); + } else if (job === "alerts") { + await sendAlerts( + container, + createUserDirectory(env), + createMailer(env), + createNotifier(env), + { + ...(env.ALERT_EXPIRY_WARNING_DAYS !== undefined + ? { expiryWarningDays: env.ALERT_EXPIRY_WARNING_DAYS } + : {}), + ...(env.ALERT_BIG_CHANGE_PERCENT !== undefined + ? { bigChangePercent: env.ALERT_BIG_CHANGE_PERCENT } + : {}), + }, + ); } else { await sendDigests( container, diff --git a/apps/worker/src/jobs/send-alerts.ts b/apps/worker/src/jobs/send-alerts.ts new file mode 100644 index 0000000..ce52ce9 --- /dev/null +++ b/apps/worker/src/jobs/send-alerts.ts @@ -0,0 +1,62 @@ +import { + deriveAlerts, + type DeriveAlertsOptions, + type Mailer, + type Notifier, + type UserDirectory, +} from "@pointup/core"; + +import type { WorkerContainer } from "../container"; +import { renderAlertsChat, renderAlertsEmail } from "../alerts-render"; + +/** + * Scheduled job: derive urgent, actionable alerts per user (expiring points, + * reached goals, large balance moves) and push them via chat (Notifier) and, + * when an email is resolvable, email. Users with no alerts are skipped — the + * point of alerts is that they only fire when something needs attention. + */ +export async function sendAlerts( + container: WorkerContainer, + directory: UserDirectory, + mailer: Mailer, + notifier: Notifier | null, + options: DeriveAlertsOptions = {}, +): Promise { + const userIds = await container.accounts.listUserIds(); + console.info(`[alerts] scanning ${userIds.length} user(s)`); + + let alertedUsers = 0; + let totalAlerts = 0; + for (const userId of userIds) { + const digest = + await container.useCases.buildPortfolioDigest.execute(userId); + if (digest.accounts.length === 0) continue; + + const alerts = deriveAlerts(digest, options); + if (alerts.length === 0) continue; + + alertedUsers += 1; + totalAlerts += alerts.length; + + if (notifier) { + await notifier + .notify(renderAlertsChat(alerts)) + .catch((error: unknown) => + console.warn(`[alerts] user=${userId} chat notify failed`, error), + ); + } + + const email = await directory.getEmail(userId).catch(() => null); + if (email) { + await mailer + .send(renderAlertsEmail(alerts, email)) + .catch((error: unknown) => + console.warn(`[alerts] user=${userId} email failed`, error), + ); + } + } + + console.info( + `[alerts] done: ${totalAlerts} alert(s) across ${alertedUsers} user(s)`, + ); +} diff --git a/docker-compose.yml b/docker-compose.yml index 5ba82ee..ba51de3 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -55,6 +55,7 @@ services: # Background jobs. Run on demand: # docker compose run --rm worker sync # docker compose run --rm worker digest + # docker compose run --rm worker alerts worker: build: context: . diff --git a/docs/bot.md b/docs/bot.md index 237716a..153040a 100644 --- a/docs/bot.md +++ b/docs/bot.md @@ -93,6 +93,27 @@ npx cdk deploy -c slackWebhookUrl=https://hooks.slack.com/services/... \ -c discordWebhookUrl=https://discord.com/api/webhooks/... ``` +## Proactive alerts + +Separate from the weekly digest, the worker's `alerts` job pushes **urgent, +actionable** items — and only when there's something to say (nothing fires for a +calm portfolio). Alerts are derived statelessly by `deriveAlerts` in the core +from the same digest read model: + +- **Expiring / expired** points (within the warning window) +- **Balance drops / jumps** past a threshold (from each account's `trend`) +- **Goal reached** — enough points to book a trip goal + +Delivery reuses the `Notifier` (Slack/Discord) and `Mailer` ports. Run it daily: + +```bash +npm run dev --workspace @pointup/worker # then: alerts +# or: docker compose run --rm worker alerts +``` + +Thresholds are tunable via `ALERT_EXPIRY_WARNING_DAYS` and +`ALERT_BIG_CHANGE_PERCENT`. In AWS the `AlertsTask` runs daily at 12:00 UTC. + ## Deploying the bot `Dockerfile.bot` builds a self-contained bundle (`node index.cjs`, port 8080, diff --git a/infra/lib/app-stack.ts b/infra/lib/app-stack.ts index ca4db82..75923b2 100644 --- a/infra/lib/app-stack.ts +++ b/infra/lib/app-stack.ts @@ -335,7 +335,48 @@ export class AppStack extends cdk.Stack { }), ); - for (const task of [syncTask, digestTask]) { + // ─── Proactive alerts task (daily) ───────────────────────────────────── + // Fires only when a user has something worth flagging (expiring points, + // reached goals, large balance moves); delivers via chat + email. + const chatWebhookEnv: Record = { + ...(ctx("slackWebhookUrl") + ? { SLACK_WEBHOOK_URL: ctx("slackWebhookUrl")! } + : {}), + ...(ctx("discordWebhookUrl") + ? { DISCORD_WEBHOOK_URL: ctx("discordWebhookUrl")! } + : {}), + }; + const alertsTask = new ecsPatterns.ScheduledFargateTask(this, "AlertsTask", { + cluster, + // Daily at 12:00 UTC. + schedule: events.Schedule.cron({ hour: "12", minute: "0" }), + subnetSelection: { subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS }, + scheduledFargateTaskImageOptions: { + image: ecs.ContainerImage.fromDockerImageAsset(workerImage), + command: ["alerts"], + cpu: 256, + memoryLimitMiB: 512, + environment: { + NODE_ENV: "production", + MAILER: digestFromEmail ? "ses" : "console", + ...(digestFromEmail ? { DIGEST_FROM_EMAIL: digestFromEmail } : {}), + ...chatWebhookEnv, + }, + secrets: workerSecrets, + logDriver: ecs.LogDrivers.awsLogs({ + streamPrefix: "worker-alerts", + logRetention: logs.RetentionDays.ONE_MONTH, + }), + }, + }); + alertsTask.taskDefinition.taskRole.addToPrincipalPolicy( + new iam.PolicyStatement({ + actions: ["ses:SendEmail", "ses:SendRawEmail"], + resources: ["*"], + }), + ); + + for (const task of [syncTask, digestTask, alertsTask]) { database.connections.allowDefaultPortFrom( task.task.securityGroups![0]!, "Worker tasks to PostgreSQL", diff --git a/packages/core/src/application/loyalty/derive-alerts.ts b/packages/core/src/application/loyalty/derive-alerts.ts new file mode 100644 index 0000000..90e933d --- /dev/null +++ b/packages/core/src/application/loyalty/derive-alerts.ts @@ -0,0 +1,118 @@ +import type { PortfolioDigestReadModel } from "./build-portfolio-digest"; +import { DEFAULT_EXPIRY_WARNING_DAYS } from "./list-expiring-accounts"; + +export type AlertType = + | "expired" + | "expiring" + | "goal-reached" + | "balance-drop" + | "balance-jump"; + +export type AlertSeverity = "warning" | "info"; + +export interface PortfolioAlert { + readonly type: AlertType; + readonly severity: AlertSeverity; + /** One-line, human-readable alert text. */ + readonly message: string; + readonly providerId?: string; + readonly goalId?: string; +} + +export interface DeriveAlertsOptions { + /** Flag accounts expiring within this many days. */ + readonly expiryWarningDays?: number; + /** Minimum |percent| change vs. the previous snapshot to flag (0.2 = 20%). */ + readonly bigChangePercent?: number; + /** Ignore changes below this many points, to avoid noise on tiny balances. */ + readonly bigChangeMinPoints?: number; +} + +const numberFormat = new Intl.NumberFormat("en-US"); + +/** + * Pure, stateless derivation of urgent, actionable alerts from the same + * portfolio digest read model the weekly summary uses. Stateless because the + * signals it needs are already in the read model: `daysUntilExpiry`, per-account + * `trend` (delta vs. the previous snapshot), and goal `achieved`. No extra + * persistence or "last alerted" bookkeeping required. + * + * Ordered warnings-first so delivery surfaces can lead with what matters. + */ +export function deriveAlerts( + digest: PortfolioDigestReadModel, + options: DeriveAlertsOptions = {}, +): PortfolioAlert[] { + const warningDays = options.expiryWarningDays ?? DEFAULT_EXPIRY_WARNING_DAYS; + const bigPercent = options.bigChangePercent ?? 0.2; + const minPoints = options.bigChangeMinPoints ?? 1000; + + const alerts: PortfolioAlert[] = []; + + for (const account of digest.accounts) { + const name = account.provider.displayName; + const days = account.daysUntilExpiry; + + if (days !== null && days < 0) { + alerts.push({ + type: "expired", + severity: "warning", + message: `${name} points may have expired — sync or check your account.`, + providerId: account.provider.id, + }); + } else if (days !== null && days <= warningDays) { + alerts.push({ + type: "expiring", + severity: "warning", + message: `${name} expires in ${days} day${days === 1 ? "" : "s"} — earn, redeem, or sync to reset the clock.`, + providerId: account.provider.id, + }); + } + + const delta = account.trend.sincePrevious; + if ( + delta && + delta.percent !== null && + Math.abs(delta.points) >= minPoints && + Math.abs(delta.percent) >= bigPercent + ) { + const pct = Math.round(Math.abs(delta.percent) * 100); + if (delta.points < 0) { + alerts.push({ + type: "balance-drop", + severity: "warning", + message: `${name} dropped ${numberFormat.format(-delta.points)} points (−${pct}%) since the last sync.`, + providerId: account.provider.id, + }); + } else { + alerts.push({ + type: "balance-jump", + severity: "info", + message: `${name} is up ${numberFormat.format(delta.points)} points (+${pct}%) since the last sync.`, + providerId: account.provider.id, + }); + } + } + } + + for (const goal of digest.goals) { + if (goal.achieved) { + alerts.push({ + type: "goal-reached", + severity: "info", + message: `Goal reached: “${goal.title}” — you have enough points to book.`, + goalId: goal.id, + }); + } + } + + // Warnings first, preserving discovery order within each severity. + return alerts + .map((alert, index) => ({ alert, index })) + .sort((a, b) => severityRank(a.alert) - severityRank(b.alert) || a.index - b.index) + .map(({ alert }) => alert); +} + +function severityRank(alert: PortfolioAlert): number { + return alert.severity === "warning" ? 0 : 1; +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 2ceafd3..3a901a4 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -20,6 +20,7 @@ export * from "./application/loyalty/get-balance-history"; export * from "./application/loyalty/record-manual-balance"; export * from "./application/loyalty/get-portfolio-summary"; export * from "./application/loyalty/build-portfolio-digest"; +export * from "./application/loyalty/derive-alerts"; export * from "./application/loyalty/export-portfolio"; export * from "./application/loyalty/import-portfolio"; export * from "./application/loyalty/balance-trend"; diff --git a/packages/core/test/derive-alerts.test.ts b/packages/core/test/derive-alerts.test.ts new file mode 100644 index 0000000..4e02f2a --- /dev/null +++ b/packages/core/test/derive-alerts.test.ts @@ -0,0 +1,145 @@ +import { describe, expect, it } from "vitest"; + +import { deriveAlerts } from "../src/application/loyalty/derive-alerts"; +import type { PortfolioDigestReadModel } from "../src/application/loyalty/build-portfolio-digest"; +import type { LoyaltyAccountReadModel } from "../src/application/loyalty/read-models"; +import type { TripGoalReadModel } from "../src/application/loyalty/create-trip-goal"; +import type { BalanceTrend } from "../src/application/loyalty/balance-trend"; + +function account( + over: { + id: string; + daysUntilExpiry?: number | null; + sincePrevious?: BalanceTrend["sincePrevious"]; + }, +): LoyaltyAccountReadModel { + return { + id: over.id, + provider: { + id: over.id, + kind: "airline", + displayName: over.id.toUpperCase(), + pointsCurrency: "miles", + estimatedCentsPerPoint: 1.3, + inactivityExpiryMonths: 18, + }, + membershipNumber: "X", + hasStoredCredential: false, + latestBalance: { points: 10_000, source: "manual", capturedAt: new Date("2026-01-01") }, + estimatedValueCents: 13_000, + trend: { + sincePrevious: over.sincePrevious ?? null, + since30Days: null, + since90Days: null, + }, + expiresAt: null, + daysUntilExpiry: over.daysUntilExpiry ?? null, + notes: null, + tags: [], + pinnedAt: null, + createdAt: new Date("2026-01-01"), + }; +} + +function goal(over: Partial & { id: string; achieved: boolean }): TripGoalReadModel { + return { + id: over.id, + title: over.title ?? "Kyoto", + targetPoints: 100_000, + targetDate: null, + accountIds: [], + status: "active", + notes: null, + currentPoints: over.achieved ? 100_000 : 40_000, + remainingPoints: over.achieved ? 0 : 60_000, + percentComplete: over.achieved ? 100 : 40, + achieved: over.achieved, + createdAt: new Date("2026-01-01"), + updatedAt: new Date("2026-01-01"), + }; +} + +function digest( + accounts: LoyaltyAccountReadModel[], + goals: TripGoalReadModel[] = [], +): PortfolioDigestReadModel { + return { + userId: "u", + summary: { + totalPoints: 0, + totalValueCents: 0, + accountCount: accounts.length, + byKind: {} as PortfolioDigestReadModel["summary"]["byKind"], + lastSyncedAt: null, + }, + accounts, + goals, + expiring: [], + }; +} + +describe("deriveAlerts", () => { + it("flags expiring and expired accounts", () => { + const alerts = deriveAlerts( + digest([ + account({ id: "delta", daysUntilExpiry: 10 }), + account({ id: "aa", daysUntilExpiry: -3 }), + account({ id: "united", daysUntilExpiry: 400 }), // outside window + account({ id: "amex", daysUntilExpiry: null }), // never expires + ]), + ); + const types = alerts.map((a) => a.type); + expect(types).toContain("expiring"); + expect(types).toContain("expired"); + expect(alerts.some((a) => a.providerId === "united")).toBe(false); + expect(alerts.some((a) => a.providerId === "amex")).toBe(false); + }); + + it("flags big balance drops (warning) and jumps (info) past the thresholds", () => { + const alerts = deriveAlerts( + digest([ + account({ id: "drop", sincePrevious: { points: -5000, percent: -0.4 } }), + account({ id: "jump", sincePrevious: { points: 8000, percent: 0.5 } }), + account({ id: "tiny", sincePrevious: { points: -50, percent: -0.9 } }), // below min points + account({ id: "small", sincePrevious: { points: 5000, percent: 0.05 } }), // below min percent + ]), + ); + const byProvider = new Map(alerts.map((a) => [a.providerId, a])); + expect(byProvider.get("drop")?.type).toBe("balance-drop"); + expect(byProvider.get("drop")?.severity).toBe("warning"); + expect(byProvider.get("jump")?.type).toBe("balance-jump"); + expect(byProvider.get("jump")?.severity).toBe("info"); + expect(byProvider.has("tiny")).toBe(false); + expect(byProvider.has("small")).toBe(false); + }); + + it("flags reached goals and orders warnings before info", () => { + const alerts = deriveAlerts( + digest( + [ + // info-severity balance jump + account({ id: "jump", sincePrevious: { points: 8000, percent: 0.5 } }), + // warning-severity expiry + account({ id: "delta", daysUntilExpiry: 10 }), + ], + [goal({ id: "g1", achieved: true }), goal({ id: "g2", achieved: false })], + ), + ); + expect(alerts.some((a) => a.type === "goal-reached" && a.goalId === "g1")).toBe(true); + expect(alerts.some((a) => a.goalId === "g2")).toBe(false); + // All warnings come before any info alert. + const firstInfo = alerts.findIndex((a) => a.severity === "info"); + const lastWarning = alerts.map((a) => a.severity).lastIndexOf("warning"); + expect(lastWarning).toBeLessThan(firstInfo); + }); + + it("returns nothing for a calm portfolio", () => { + expect(deriveAlerts(digest([account({ id: "united", daysUntilExpiry: 400 })]))).toEqual([]); + }); + + it("respects custom thresholds", () => { + const d = digest([account({ id: "x", daysUntilExpiry: 30 })]); + expect(deriveAlerts(d, { expiryWarningDays: 14 })).toEqual([]); + expect(deriveAlerts(d, { expiryWarningDays: 60 })).toHaveLength(1); + }); +});