Feature: Real-Time Rules & Anomaly Alerts (Stream Processing over Kafka/RabbitMQ)

### Summary

Add a **Rules & Alerts** subsystem that evaluates incoming budget/expense events in real time and generates notifications for threshold breaches (e.g., budget overspend, burn-rate spike) and **statistical anomalies** (duplicate/rare merchant, outlier amounts). Alerts are delivered via WebSockets (immediate), stored for audit, and queryable via REST/GraphQL/gRPC. Rules are user-configurable and can be tested before activation.

---

### Motivation

* Today we persist budgets/expenses and support search/notifications, but there’s no **proactive** protection.
* A rules engine + lightweight anomaly detection reduces surprise overruns and flags suspicious transactions.
* Leverages existing infra: **Kafka/RabbitMQ**, **Redis**, **Elasticsearch**, **WebSockets**, **GraphQL/gRPC**.

---

### Goals

1. **Rule management API** (CRUD, enable/disable, dry-run test).
2. **Stream processor** consumes `expenses.created` / `transactions.created` and emits `alerts.created`.
3. **Alert delivery** via WebSockets; persistence in MongoDB; index in Elasticsearch for querying.
4. **Anomaly detection v1**: z-score over rolling windows + duplicate/similarity checks.
5. **GraphQL + gRPC** read APIs; REST parity.
6. **CLI** helpers to seed common rules and list recent alerts.

**Non-Goals (v1):** complex ML models, cross-user correlation, email/SMS channels (can be follow-ups).

---

### Acceptance Criteria

* Creating a threshold rule (“**Budget > 90% before period end**”) triggers an alert on next qualifying event.
* Anomaly engine flags an expense when amount is **> 3σ** over user/category rolling mean (min history N=20).
* Web UI (demo frontend) receives a `alerts.created` WebSocket broadcast within **< 1s p95** of event ingestion.
* Alerts are persisted, **idempotent** (dedup by `(userId, ruleId, expenseId, windowKey)`), and **ack/resolvable**.
* Queries by time range, severity, rule, budget, category return in **< 200ms p95** (backed by ES).
* GraphQL returns alerts with pagination and facets; gRPC returns stream or unary list.

---

### Rule Types (v1)

* **Threshold:** budget utilization %, category daily cap, single expense amount cap.
* **Burn-rate:** projected end-of-period spend exceeds limit given current slope (EWMA).
* **Duplicate:** same amount + merchant within short window; or same invoiceId.
* **Outlier:** z-score or MAD over past K expenses (per user/category).
* **Missing recurring:** expected monthly subscription not observed by due date (+/- grace).

---

### Data Model (MongoDB)

* `rules`: `{ _id, userId, name, type, params, enabled, severity, channels, createdAt, updatedAt }`
* `alerts`: `{ _id, userId, ruleId, budgetId?, expenseId?, type, message, details, severity, status: "open|ack|resolved", createdAt }`
* `metrics_rollup` (optional): `{ userId, category, window, mean, stddev, count, updatedAt }`

**Elasticsearch index:** `alerts-*` with fields for time, type, severity, ruleId, budgetId, category, merchant.

---

### REST API (proposed)

```
# Rules
POST /api/rules
GET /api/rules
GET /api/rules/:id
PUT /api/rules/:id
DELETE /api/rules/:id
POST /api/rules/:id/test # dry-run against last N expenses

# Alerts
GET /api/alerts?from=&to=&status=&severity=&ruleId=&budgetId=&category=
POST /api/alerts/:id/ack
POST /api/alerts/:id/resolve
```

**Example rule payload**

```json
{
"name": "Budget 90% Early Warning",
"type": "threshold.budget_utilization",
"severity": "high",
"enabled": true,
"params": { "percent": 90, "beforeDays": 5, "scopes": ["monthly"] },
"channels": ["websocket"]
}
```

**Example alert**

```json
{
"userId": "u_123",
"ruleId": "r_abc",
"type": "threshold.budget_utilization",
"severity": "high",
"message": "Budget 'Groceries' at 92% with 6 days left.",
"details": { "budgetId": "b_1", "utilization": 0.92, "periodEndsAt": "2025-08-31" },
"status": "open",
"createdAt": "2025-08-17T02:11:00Z"
}
```

---

### GraphQL (proposed)

```graphql
type Rule {
id: ID!
name: String!
type: String!
enabled: Boolean!
severity: String!
params: JSON
createdAt: String!
updatedAt: String!
}

type Alert {
id: ID!
userId: ID!
ruleId: ID!
type: String!
severity: String!
message: String!
details: JSON
status: String!
createdAt: String!
}

type AlertPage {
items: [Alert!]!
nextCursor: String
}

input AlertFilter {
from: String
to: String
status: String
severity: String
ruleId: ID
budgetId: ID
category: String
}

type Query {
rules: [Rule!]!
alertFeed(filter: AlertFilter, cursor: String, limit: Int = 50): AlertPage!
}

type Mutation {
createRule(name: String!, type: String!, severity: String!, params: JSON, enabled: Boolean): Rule!
updateRule(id: ID!, name: String, params: JSON, enabled: Boolean, severity: String): Rule!
deleteRule(id: ID!): Boolean!
ackAlert(id: ID!): Alert!
resolveAlert(id: ID!): Alert!
}
```

---

### gRPC (proto sketch)

```proto
syntax = "proto3";
package alerts.v1;

message Alert {
string id = 1;
string userId = 2;
string ruleId = 3;
string type = 4;
string severity = 5;
string message = 6;
string createdAt = 7;
string status = 8;
string detailsJson = 9;
}

message ListAlertsRequest { string userId = 1; string from = 2; string to = 3; }
message ListAlertsResponse { repeated Alert items = 1; }

service Alerts {
rpc ListAlerts(ListAlertsRequest) returns (ListAlertsResponse);
rpc AckAlert(Alert) returns (Alert);
rpc ResolveAlert(Alert) returns (Alert);
}
```

---

### Event Topics (Kafka/RabbitMQ)

* **Input:** `expenses.created`, `transactions.created`, `budgets.updated`, `schedules.tick.daily`
* **Output:** `alerts.created` (consumed by WebSocket broadcaster + ES ingester)

Message envelope:

```json
{
"event": "expenses.created",
"ts": "2025-08-17T02:10:00Z",
"userId": "u_123",
"payload": { "expenseId": "e_9", "budgetId": "b_1", "amount": 129.99, "category": "Dining", "merchant": "Sushi Zen" }
}
```

---

### Processing Logic (v1)

* **Threshold rules:** evaluate deterministic predicates (e.g., utilization = spent/limit).
* **Outlier rules:** maintain rolling stats per `(userId, category)` in Redis; fallback to Mongo rollups.
* **Duplicate detection:** key = `hash(userId, amount, merchant, ±window)` in Redis with short TTL; if exists → alert.
* **Burn-rate:** EWMA of daily spend; project to period end; compare to limit.
* **Idempotency:** dedup store keyed by `(userId, ruleId, expenseId|dayBucket)` with TTL.

---

### Security & Privacy

* Enforce JWT on all rule/alert endpoints.
* Per-user isolation in Kafka consumer paths; never cross-tenant join.
* PII-light payloads in events (no card PANs; merchant names allowed).

---

### Performance Targets

* Stream ingest → alert broadcast **< 1s p95**.
* Rule evaluation **< 20ms** per event average under 100 rps.
* Query (ES) p95 **< 200ms** for last 30 days, 100k alerts.

---

### Observability

* Prometheus counters: `alerts_created_total{type,severity}`, `rules_evaluated_total{result}`, `anomaly_flags_total`.
* Histograms: end-to-end latency, evaluation latency.
* Grafana dashboards + alerting on spikes.

---

### CLI Additions

```
# Seed common rules
budget-manager rules seed-common

# List recent alerts
budget-manager alerts list --since 24h --severity high
```

---

### Rollout Plan

1. **Phase 1:** Rules CRUD + threshold engine + WebSocket delivery + persistence/ES indexing.
2. **Phase 2:** Anomaly rules (z-score, duplicate), burn-rate projection, CLI helpers.
3. **Phase 3:** GraphQL/gRPC endpoints, dashboards, dry-run test API, demo-frontend widgets.

---

### Tasks

**Backend / API**

* [ ] `/api/rules` CRUD, validation, auth.
* [ ] `/api/rules/:id/test` dry-run against last N expenses.
* [ ] `/api/alerts` listing + ack/resolve; ES integration.

**Stream Worker**

* [ ] Kafka/RabbitMQ consumer(s) for `expenses.created` etc.
* [ ] Rule registry + executor; Redis caches; idempotency keys.
* [ ] Emit `alerts.created`; persist + broadcast.

**Search / Storage**

* [ ] Mongo schemas (`rules`, `alerts`); indexes.
* [ ] ES index template + ingestion pipeline.

**Realtime**

* [ ] WebSocket channel `alerts` with auth scopes.
* [ ] Broadcast on `alerts.created`.

**GraphQL / gRPC**

* [ ] Schema, resolvers, pagination; proto + server stubs.

**Infra / CI**

* [ ] Feature flag `alerts_enabled`.
* [ ] Prometheus/Grafana dashboards; alerting rules.
* [ ] Load tests for 100 rps ingest.

**Docs**

* [ ] Update `README`, `openapi.yaml`, GraphQL docs.
* [ ] Examples: rule payloads, alert objects, CLI usage.

---

### Open Questions

* Default anomaly thresholds (z=3 vs adaptive by category)?
* Dedup window for duplicates (5–15 min by default)?
* Should burn-rate consider seasonality (weekends) in v1 or keep linear/EWMA?
* Retention policy for alerts in Mongo/ES (e.g., 180 days)?

---