commit 8815c20c1315c08ad7a68d5a5ed3983c952dcf16 Author: thuanle Date: Mon Apr 27 18:16:57 2026 +0700 Initial commit: email-openclaw bridge v1 Full pipeline: IMAP ingress -> OpenClaw dispatch -> callback -> SMTP reply. SQLite stateful storage with idempotency, threading, and retry logic. Co-Authored-By: Claude Opus 4.7 diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..5a88668 --- /dev/null +++ b/.env.example @@ -0,0 +1,24 @@ +# === IMAP === +IMAP_HOST=imap.example.com +IMAP_PORT=993 +IMAP_USER=bridge@example.com +IMAP_PASS=changeme + +# === SMTP === +SMTP_HOST=smtp.example.com +SMTP_PORT=587 +SMTP_USER=bridge@example.com +SMTP_PASS=changeme + +# === OpenClaw === +OPENCLAW_URL=https://api.openclaw.example.com +OPENCLAW_API_KEY= + +# === Bridge === +BRIDGE_CALLBACK_TOKEN=a-strong-random-token +SYSTEM_EMAIL=bridge@example.com +WHITELIST_EMAILS=user1@example.com,user2@example.com + +# === Optional === +IMAP_PROXY_URL= +LISTEN_ADDR=:8080 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..31ae8a5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +# Environment +.env + +# Database +data/*.db +data/*.db-wal +data/*.db-shm + +# Go +/bin/ +/tmp/ + +# IDE +.idea/ +.vscode/ +*.swp diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..66cd1ae --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,70 @@ +# CLAUDE.md + +Behavioral guidelines to reduce common LLM coding mistakes. Merge with project-specific instructions as needed. + +**Tradeoff:** These guidelines bias toward caution over speed. For trivial tasks, use judgment. + +## 1. Think Before Coding + +**Don't assume. Don't hide confusion. Surface tradeoffs.** + +Before implementing: + +- State your assumptions explicitly. If uncertain, ask. +- If multiple interpretations exist, present them - don't pick silently. +- If a simpler approach exists, say so. Push back when warranted. +- If something is unclear, stop. Name what's confusing. Ask. + +## 2. Simplicity First + +**Minimum code that solves the problem. Nothing speculative.** + +- No features beyond what was asked. +- No abstractions for single-use code. +- No "flexibility" or "configurability" that wasn't requested. +- No error handling for impossible scenarios. +- If you write 200 lines and it could be 50, rewrite it. + +Ask yourself: "Would a senior engineer say this is overcomplicated?" If yes, simplify. + +## 3. Surgical Changes + +**Touch only what you must. Clean up only your own mess.** + +When editing existing code: + +- Don't "improve" adjacent code, comments, or formatting. +- Don't refactor things that aren't broken. +- Match existing style, even if you'd do it differently. +- If you notice unrelated dead code, mention it - don't delete it. + +When your changes create orphans: + +- Remove imports/variables/functions that YOUR changes made unused. +- Don't remove pre-existing dead code unless asked. + +The test: Every changed line should trace directly to the user's request. + +## 4. Goal-Driven Execution + +**Define success criteria. Loop until verified.** + +Transform tasks into verifiable goals: + +- "Add validation" → "Write tests for invalid inputs, then make them pass" +- "Fix the bug" → "Write a test that reproduces it, then make it pass" +- "Refactor X" → "Ensure tests pass before and after" + +For multi-step tasks, state a brief plan: + +``` +1. [Step] → verify: [check] +2. [Step] → verify: [check] +3. [Step] → verify: [check] +``` + +Strong success criteria let you loop independently. Weak criteria ("make it work") require constant clarification. + +--- + +**These guidelines are working if:** fewer unnecessary changes in diffs, fewer rewrites due to overcomplication, and clarifying questions come before implementation rather than after mistakes. diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..fc68716 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,29 @@ +# ---- Build stage ---- +FROM golang:1.23-alpine AS builder + +WORKDIR /src + +# Cache module downloads. +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . + +# CGO disabled — using modernc.org/sqlite (pure Go). +RUN CGO_ENABLED=0 go build -o /app/bridge ./cmd/bridge + +# ---- Runtime stage ---- +FROM alpine:3.20 + +RUN apk add --no-cache ca-certificates tzdata + +WORKDIR /app + +COPY --from=builder /app/bridge . + +# Ensure data directory exists for SQLite volume mount. +RUN mkdir -p /app/data + +EXPOSE 8080 + +ENTRYPOINT ["./bridge"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..3ca5bd9 --- /dev/null +++ b/README.md @@ -0,0 +1,105 @@ +# Email-OpenClaw Bridge + +Tự động xử lý email bằng OpenClaw AI theo luồng: + +``` +Ingress (IMAP) → Dispatch (OpenClaw) → Callback → Egress (SMTP reply) +``` + +## Tính năng + +- **IMAP IDLE** — theo dõi mailbox realtime, nhận email mới tức thì +- **Threading** — giữ nguyên email thread qua `In-Reply-To` / `References` +- **Idempotency** — chống duplicate theo `message_id` +- **Anti-loop & Whitelist** — bỏ qua email từ chính hệ thống, chỉ xử lý sender được phép +- **Retry** — tối đa 3 lần với backoff 1s → 5s → 15s cho cả OpenClaw và SMTP +- **Stateful** — SQLite lưu trạng thái pipeline, không mất dữ liệu khi restart + +## Kiến trúc + +``` +cmd/bridge/main.go Entrypoint +internal/ +├── config/ Đọc cấu hình từ .env +├── database/ SQLite + GORM, auto-migrate +├── logging/ Structured JSON logging (slog) +├── api/ HTTP server (Gin): health checks, webhook callback +├── mail/ IMAP ingress + SMTP egress +└── ai_client/ OpenClaw HTTP dispatch +``` + +## Yêu cầu + +- Go 1.23+ +- Docker & Docker Compose (cho deploy) + +## Quickstart + +### 1. Cấu hình + +```bash +cp .env.example .env +# Sửa .env với thông tin IMAP/SMTP/OpenClaw thực tế +``` + +### 2. Chạy local (development) + +```bash +go run ./cmd/bridge +``` + +### 3. Chạy bằng Docker + +```bash +docker compose up -d +docker logs -f claw-email-bridge +``` + +## Cấu hình (.env) + +| Biến | Bắt buộc | Mô tả | +|---|:---:|---| +| `IMAP_HOST` | ✅ | IMAP server hostname | +| `IMAP_PORT` | ✅ | IMAP port (thường `993`) | +| `IMAP_USER` | ✅ | IMAP username | +| `IMAP_PASS` | ✅ | IMAP password | +| `SMTP_HOST` | ✅ | SMTP server hostname | +| `SMTP_PORT` | ✅ | SMTP port (thường `587`) | +| `SMTP_USER` | ✅ | SMTP username | +| `SMTP_PASS` | ✅ | SMTP password | +| `OPENCLAW_URL` | ✅ | OpenClaw API endpoint | +| `OPENCLAW_API_KEY` | | OpenClaw API key (nếu cần) | +| `BRIDGE_CALLBACK_TOKEN` | ✅ | Token xác thực webhook callback | +| `SYSTEM_EMAIL` | ✅ | Email của hệ thống (dùng cho anti-loop) | +| `WHITELIST_EMAILS` | | Danh sách email được phép, phân cách bằng dấu phẩy | +| `IMAP_PROXY_URL` | | SOCKS5 proxy cho IMAP (vd: `socks5://user:pass@host:port`) | +| `LISTEN_ADDR` | | Địa chỉ HTTP server (mặc định `:8080`) | + +## Health Checks + +| Endpoint | Mục đích | +|---|---| +| `GET /healthz` | Liveness — process còn sống | +| `GET /readyz` | Readiness — DB writable + config hợp lệ | + +## Pipeline trạng thái + +``` +RECEIVED → AI_PROCESSING → CALLBACK_DONE → COMPLETED + ↓ ↓ ↓ + IGNORED FAILED SMTP_RETRYING → FAILED +``` + +## Công nghệ + +| Thành phần | Thư viện | +|---|---| +| HTTP Server | [gin-gonic/gin](https://github.com/gin-gonic/gin) | +| Database | [GORM](https://gorm.io) + [modernc.org/sqlite](https://pkg.go.dev/modernc.org/sqlite) (pure Go, no CGO) | +| IMAP | [emersion/go-imap](https://github.com/emersion/go-imap) | +| Logging | [log/slog](https://pkg.go.dev/log/slog) (stdlib) | +| Config | [joho/godotenv](https://github.com/joho/godotenv) | + +## License + +Private — internal use only. diff --git a/checkpoints/001-feedback.md b/checkpoints/001-feedback.md new file mode 100644 index 0000000..524539c --- /dev/null +++ b/checkpoints/001-feedback.md @@ -0,0 +1,17 @@ +# Checkpoint 001 — Feedback kết quả kiểm tra + +## Kết luận +✅ **Đạt** theo tiêu chí checkpoint. + +## Bằng chứng +- Verify command: `go build ./...` chạy thành công. +- Cấu trúc scaffold khớp mô tả checkpoint: + - Có `cmd/bridge/main.go`, `internal/config`, `internal/database`, `internal/logging`, `internal/api`. + - Có `.env.example`, `Dockerfile`, `docker-compose.yml`, `go.mod`, `go.sum`. +- Các phần cốt lõi scaffold hiện diện: + - SQLite + WAL + auto-migrate. + - `/healthz`, `/readyz`. + - Validate config bắt buộc. + +## Lưu ý (không chặn checkpoint) +- README mô tả tính năng pipeline/retry/threading khá đầy đủ, nhưng tại checkpoint 001 phần mail/ai vẫn chủ yếu ở mức scaffold/stub. Đây không mâu thuẫn với checkpoint 001, chỉ cần hiểu đúng phạm vi. diff --git a/checkpoints/001-scaffold.md b/checkpoints/001-scaffold.md new file mode 100644 index 0000000..b38d526 --- /dev/null +++ b/checkpoints/001-scaffold.md @@ -0,0 +1,91 @@ +# Checkpoint 001 — Project Scaffold + +**Ngày:** 2026-04-24 +**Trạng thái:** ✅ Hoàn thành +**Verify:** `go build ./...` thành công, không lỗi + +--- + +## Việc đã làm + +### 1. Khởi tạo Go module +``` +go mod init github.com/thuanle/claw-email +``` + +### 2. Tạo cấu trúc thư mục + +``` +claw-email/ +├── cmd/bridge/main.go # Entrypoint +├── internal/ +│ ├── config/config.go # Đọc .env, validate required vars +│ ├── database/ +│ │ ├── database.go # Open SQLite + auto-migrate + health check +│ │ └── models.go # Task model (đầy đủ schema từ requirements) +│ ├── logging/logger.go # Structured JSON logging (slog) +│ ├── api/router.go # Gin router: /healthz, /readyz +│ ├── mail/mail.go # [stub] IMAP ingress + SMTP egress +│ └── ai_client/client.go # [stub] OpenClaw HTTP dispatch +├── .env.example +├── .gitignore +├── Dockerfile # Multi-stage, CGO_ENABLED=0 +├── docker-compose.yml # Named volume cho SQLite +├── go.mod / go.sum +└── requirements.md +``` + +### 3. Kết nối SQLite + auto-migrate bảng `tasks` +- Driver: `github.com/glebarez/sqlite` (pure Go, wraps `modernc.org/sqlite`) +- WAL mode enabled +- Auto-migrate bảng `tasks` khi khởi động +- Health check: `SELECT 1` + +### 4. Module config (.env) +- Dùng `github.com/joho/godotenv` load `.env` +- Validate tất cả biến bắt buộc, trả lỗi rõ ràng nếu thiếu +- Parse `WHITELIST_EMAILS` (comma-separated) +- Default `LISTEN_ADDR=:8080` + +### 5. Structured Logging +- Dùng `log/slog` (Go stdlib 1.21+), JSON handler ra stdout +- Helper `TaskLogger(taskUUID, threadID, messageID)` — mọi log liên quan task đều có context + +### 6. API endpoints (Gin) +- `GET /healthz` → `200 {"status":"ok"}` (liveness) +- `GET /readyz` → kiểm tra DB writable + config token (readiness) + +### 7. Docker +- **Dockerfile**: Multi-stage, `golang:1.23-alpine` → `alpine:3.20`, `CGO_ENABLED=0` +- **docker-compose.yml**: Named volume `bridge-data:/app/data`, `env_file: .env` + +--- + +## Quyết định thiết kế + +| Quyết định | Lý do | +|---|---| +| `glebarez/sqlite` thay vì `gorm.io/driver/sqlite` | Pure Go, không cần CGO → Docker build đơn giản | +| `log/slog` thay vì zap/logrus | Stdlib, zero dependency, đủ dùng | +| `gin-gonic/gin` | Đúng requirements, nhẹ, phổ biến | +| WAL mode SQLite | Tăng throughput đọc concurrent | +| `internal/` layout | Chuẩn Go, ngăn import từ bên ngoài | + +## Dependencies (go.mod) + +``` +github.com/gin-gonic/gin v1.12.0 +github.com/glebarez/sqlite v1.11.0 (→ modernc.org/sqlite) +github.com/joho/godotenv v1.5.1 +gorm.io/gorm v1.31.1 +``` + +--- + +## Tiếp theo (chưa implement) + +1. IMAP Ingress — IDLE, anti-loop, whitelist, idempotency, threading +2. OpenClaw Dispatch — HTTP POST, retry 3 lần (1s/5s/15s) +3. Callback handler — webhook xác thực token, cập nhật task +4. SMTP Egress — gửi reply, In-Reply-To/References, retry +5. Integration tests — 6 scenarios từ requirements diff --git a/checkpoints/002-feedback.md b/checkpoints/002-feedback.md new file mode 100644 index 0000000..98afb7c --- /dev/null +++ b/checkpoints/002-feedback.md @@ -0,0 +1,15 @@ +# Checkpoint 002 — Feedback kết quả kiểm tra + +## Kết luận +✅ **Đạt** theo tiêu chí checkpoint. + +## Bằng chứng +- Verify command: `go build ./...` chạy thành công. +- Khớp claims chính của checkpoint 002: + - Có `internal/database/repository.go` với các helper: `FindByMessageID`, `FindLatestByThreadID`, `FindByTaskUUID`, `ThreadHistory`, `CreateTask`, `UpdateTask`. + - Có `internal/mail/imap.go` với: IMAP IDLE loop, reconnect sau 10s, fetch UNSEEN khi startup, anti-loop, whitelist, idempotency, threading theo `In-Reply-To`, save task `RECEIVED`, mark `\\Seen`. + - `cmd/bridge/main.go` chạy IMAP watcher + HTTP server concurrently, có graceful shutdown. + - Stub cũ `internal/mail/mail.go` đã được xóa. + +## Lưu ý (không chặn checkpoint) +- Threading hiện parse `In-Reply-To`; chưa thấy fallback rõ ràng theo `References` như requirement đầy đủ. Nên bổ sung ở checkpoint tiếp theo. diff --git a/checkpoints/002-imap-ingress.md b/checkpoints/002-imap-ingress.md new file mode 100644 index 0000000..268d471 --- /dev/null +++ b/checkpoints/002-imap-ingress.md @@ -0,0 +1,82 @@ +# Checkpoint 002 — IMAP Ingress + +**Ngày:** 2026-04-24 +**Trạng thái:** ✅ Hoàn thành +**Verify:** `go build ./...` thành công + +--- + +## Việc đã làm + +### 1. Task Repository (`internal/database/repository.go`) +Thêm các helper function cho CRUD operations: +- `FindByMessageID` — lookup cho idempotency check +- `FindLatestByThreadID` — tìm task COMPLETED gần nhất trong thread +- `FindByTaskUUID` — lookup cho callback handler +- `ThreadHistory` — lấy toàn bộ history thread (cho OpenClaw context) +- `CreateTask`, `UpdateTask` — CRUD cơ bản + +### 2. IMAP Watcher (`internal/mail/imap.go`) +Module chính cho Ingress, bao gồm: + +**Kết nối & IDLE:** +- `DialTLS` → Login → Select INBOX +- IDLE loop với timer 28 phút (RFC recommends < 29 min) +- Auto-reconnect sau 10s khi mất kết nối +- Graceful shutdown qua `context.Context` +- `UnilateralDataHandler` nhận notification khi có email mới + +**Xử lý email (đúng 5 bước theo requirements 4.1):** +1. **Anti-loop**: sender == `SYSTEM_EMAIL` → `IGNORED` +2. **Whitelist**: sender không trong danh sách → `IGNORED` (empty whitelist = cho phép tất cả) +3. **Idempotency**: `message_id` đã tồn tại → skip +4. **Threading**: parse `In-Reply-To` header → tìm parent task → kế thừa `thread_id` +5. **Save**: tạo task `status = RECEIVED` + +**Extras:** +- Fetch UNSEEN messages on startup (xử lý email đến trong lúc offline) +- Mark `\Seen` sau khi xử lý xong +- Lưu task IGNORED vào DB cho audit +- Parse body parts qua `go-message` để lấy plain text +- Hook `OnReceived` cho dispatch (chưa wire) + +### 3. Entrypoint (`cmd/bridge/main.go`) +- Chạy IMAP watcher + HTTP server concurrently +- Graceful shutdown qua SIGINT/SIGTERM +- `sync.WaitGroup` đợi cả hai goroutine + +--- + +## Dependencies mới + +``` +github.com/emersion/go-imap/v2 v2.0.0-beta.8 +github.com/emersion/go-message v0.18.2 +github.com/google/uuid (for task_uuid generation) +``` + +## Quyết định thiết kế + +| Quyết định | Lý do | +|---|---| +| go-imap v2 thay vì v1 | IDLE tích hợp sẵn, API hiện đại hơn, dự án mới nên dùng v2 | +| IDLE timer 28 min | RFC recommends < 29 min, close + reopen IDLE để tránh server timeout | +| Fetch UNSEEN on startup | Xử lý email đến lúc bridge offline | +| Empty whitelist = allow all | UX tốt hơn: không cấu hình whitelist thì mặc định chấp nhận mọi sender | +| Lưu IGNORED vào DB | Auditability: biết email nào bị bỏ qua và lý do | +| `UnilateralDataHandler` | Nhận notification realtime khi mailbox thay đổi (không cần polling) | + +--- + +## Files thay đổi + +- **Mới:** `internal/database/repository.go` +- **Mới:** `internal/mail/imap.go` +- **Cập nhật:** `cmd/bridge/main.go` (IMAP watcher + graceful shutdown) +- **Xóa:** `internal/mail/mail.go` (stub cũ) + +## Tiếp theo + +1. **OpenClaw Dispatch** — wire `OnReceived` → HTTP POST to OpenClaw +2. **Callback handler** — `POST /callback` + token auth +3. **SMTP Egress** — gửi reply với In-Reply-To/References diff --git a/checkpoints/003-feedback.md b/checkpoints/003-feedback.md new file mode 100644 index 0000000..4f999cc --- /dev/null +++ b/checkpoints/003-feedback.md @@ -0,0 +1,14 @@ +# Checkpoint 003 — Feedback kết quả kiểm tra + +## Kết luận +✅ **Đạt** theo tiêu chí checkpoint. + +## Bằng chứng +- Verify command: `go build ./...` chạy thành công. +- `internal/ai_client/client.go` đã có dispatch payload đúng spec: `input`, `session_id`, `history`, `callback_url`, `metadata.task_uuid`. +- Retry OpenClaw đúng 3 lần với backoff `1s -> 5s -> 15s`, cập nhật `attempt_openclaw`, hết retry thì `status=FAILED` + `last_error`. +- HTTP client đi direct (`Transport.Proxy=nil`), timeout 30s, bearer token optional. +- `cmd/bridge/main.go` đã wire `watcher.OnReceived -> go dispatcher.Dispatch(task)`. + +## Lưu ý (không chặn checkpoint) +- `callback_url` hiện dùng `http:///callback`; khi deploy qua domain/public endpoint có thể cần biến cấu hình base URL riêng. diff --git a/checkpoints/003-openclaw-dispatch.md b/checkpoints/003-openclaw-dispatch.md new file mode 100644 index 0000000..da2e43c --- /dev/null +++ b/checkpoints/003-openclaw-dispatch.md @@ -0,0 +1,45 @@ +# Checkpoint 003 — OpenClaw Dispatch + +**Ngày:** 2026-04-24 +**Trạng thái:** ✅ Hoàn thành +**Verify:** `go build ./...` thành công + +--- + +## Việc đã làm + +### OpenClaw Dispatcher (`internal/ai_client/client.go`) + +**Chức năng chính:** +- Gửi HTTP POST đến OpenClaw API với payload theo spec: + - `input` = body email + - `session_id` = thread_id + - `history` = các cặp user/assistant từ tasks COMPLETED cùng thread + - `callback_url` = endpoint nhận kết quả + - `metadata.task_uuid` = ID để callback tìm lại task + +**Retry logic (đúng requirements 4.2):** +- Tối đa 3 lần, backoff: 1s → 5s → 15s +- Mỗi lần retry tăng `attempt_openclaw` +- Hết retry: `status = FAILED`, ghi `last_error` + +**HTTP Client:** +- `Transport.Proxy = nil` — direct connection, không kế thừa proxy hệ thống +- Timeout 30s +- Bearer token auth (nếu `OPENCLAW_API_KEY` có giá trị) + +### Wiring (`cmd/bridge/main.go`) +- `watcher.OnReceived` → `go dispatcher.Dispatch(task)` (async) +- Pipeline hoàn chỉnh: IMAP → RECEIVED → AI_PROCESSING → (chờ callback) + +--- + +## Files thay đổi + +- **Rewrite:** `internal/ai_client/client.go` (từ stub → full implementation) +- **Cập nhật:** `cmd/bridge/main.go` (wire dispatcher) + +## Tiếp theo + +1. **Callback handler** — `POST /callback` + token auth + cập nhật `ai_response` +2. **SMTP Egress** — gửi reply với In-Reply-To/References + retry diff --git a/checkpoints/004-callback-handler.md b/checkpoints/004-callback-handler.md new file mode 100644 index 0000000..816ce66 --- /dev/null +++ b/checkpoints/004-callback-handler.md @@ -0,0 +1,48 @@ +# Checkpoint 004 — Callback Handler + +**Ngày:** 2026-04-24 +**Trạng thái:** ✅ Hoàn thành +**Verify:** `go build ./...` thành công + +--- + +## Việc đã làm + +### Callback Handler (`internal/api/router.go`) + +**Endpoint:** `POST /callback` + +**Xác thực (requirements 5.2):** +- Middleware `tokenAuthMiddleware` kiểm tra header `X-Bridge-Token` +- Thiếu hoặc sai token → `401 Unauthorized` +- Log warning kèm `remote_addr` + +**Xử lý callback (requirements 4.3 + 5.1):** +- Parse payload: `metadata.task_uuid` (required), `result` (required), `status` +- Tìm task theo `task_uuid` → 404 nếu không tìm thấy +- **Idempotent:** task đã `COMPLETED` → trả `200 {"status":"already_completed"}`, không thay đổi gì +- Cập nhật `ai_response = result`, `status = CALLBACK_DONE` +- Gọi async `onDone(task)` sau khi trả response (hook cho SMTP egress) + +**API Signature thay đổi:** +- `NewRouter(db, callbackToken, onDone)` — thêm param `OnCallbackDone` callback +- `main.go` truyền `nil` tạm (sẽ wire SMTP egress ở bước sau) + +--- + +## Files thay đổi + +- **Rewrite:** `internal/api/router.go` (thêm callback handler + token auth middleware) +- **Cập nhật:** `cmd/bridge/main.go` (cập nhật signature `NewRouter`) + +## Pipeline hiện tại + +``` +IMAP IDLE → RECEIVED → AI_PROCESSING → (OpenClaw callback) → CALLBACK_DONE → ??? + ↓ + [chờ SMTP egress] +``` + +## Tiếp theo + +1. **SMTP Egress** — gửi reply email với In-Reply-To/References + retry 3 lần diff --git a/checkpoints/004-feedback.md b/checkpoints/004-feedback.md new file mode 100644 index 0000000..4335c5f --- /dev/null +++ b/checkpoints/004-feedback.md @@ -0,0 +1,18 @@ +# Checkpoint 004 — Feedback kết quả kiểm tra + +## Kết luận +✅ **Đạt** theo tiêu chí checkpoint. + +## Bằng chứng +- Verify command: `go build ./...` chạy thành công. +- `internal/api/router.go` có `POST /callback`. +- Có middleware kiểm tra `X-Bridge-Token`; thiếu/sai token trả `401 Unauthorized` và log warning kèm `remote_addr`. +- Callback parse payload với `metadata.task_uuid` + `result` là required. +- Lookup task theo `task_uuid`; không thấy trả `404`. +- Idempotent path: task đã `COMPLETED` trả `200 {"status":"already_completed"}`. +- Callback hợp lệ cập nhật `ai_response`, `status=CALLBACK_DONE`. +- `NewRouter(db, callbackToken, onDone)` đã đổi signature; `cmd/bridge/main.go` truyền `nil` tạm. + +## Lưu ý (không chặn checkpoint) +- Hiện `onDone` chưa wire sang SMTP egress (đúng như checkpoint ghi “bước sau”). +- Trường `status` từ payload hiện chưa được dùng trong logic callback. diff --git a/checkpoints/005-feedback.md b/checkpoints/005-feedback.md new file mode 100644 index 0000000..5e78c46 --- /dev/null +++ b/checkpoints/005-feedback.md @@ -0,0 +1,14 @@ +# Checkpoint 005 — Feedback kết quả kiểm tra + +## Kết luận +✅ **Đạt** theo tiêu chí checkpoint. + +## Bằng chứng +- Verify command: `go build ./...` chạy thành công. +- Có `internal/mail/smtp.go` với gửi reply SMTP và retry 3 lần theo backoff `1s -> 5s -> 15s`. +- Mỗi lần retry tăng `attempt_smtp`, set `status=SMTP_RETRYING`; hết retry set `status=FAILED` + `last_error`; thành công set `status=COMPLETED`. +- Email reply có threading headers `In-Reply-To` và `References`, body lấy từ `task.AIResponse`. +- `cmd/bridge/main.go` đã wire SMTP egress qua `api.NewRouter(..., smtpSender.SendReplyFunc())`. + +## Lưu ý (không chặn checkpoint) +- Header threading đang dùng `task.MessageID` trực tiếp; một số hệ thống mail đôi khi yêu cầu định dạng `` nhất quán để thread ổn định hơn. diff --git a/checkpoints/005-smtp-egress.md b/checkpoints/005-smtp-egress.md new file mode 100644 index 0000000..6a4b629 --- /dev/null +++ b/checkpoints/005-smtp-egress.md @@ -0,0 +1,61 @@ +# Checkpoint 005 — SMTP Egress + +**Ngày:** 2026-04-24 +**Trạng thái:** ✅ Hoàn thành +**Verify:** `go build ./...` thành công + +--- + +## Việc đã làm + +### SMTP Sender (`internal/mail/smtp.go`) + +**Gửi reply email (requirements 4.3):** +- Subject: thêm prefix `Re:` nếu chưa có +- Body: `task.AIResponse` từ OpenClaw callback +- Threading headers: + - `In-Reply-To: ` + - `References: ` +- MIME: `text/plain; charset=UTF-8` + +**Retry logic (đúng requirements):** +- Tối đa 3 lần, backoff: 1s → 5s → 15s +- Mỗi lần retry: tăng `attempt_smtp`, status = `SMTP_RETRYING` +- Hết retry: `status = FAILED`, ghi `last_error` +- Thành công: `status = COMPLETED` + +**SMTP delivery:** +- `net/smtp.SendMail` với `PlainAuth` +- Địa chỉ: `SMTP_HOST:SMTP_PORT` +- From: `SYSTEM_EMAIL` + +### Wiring (`cmd/bridge/main.go`) +- `smtpSender.SendReplyFunc()` → `api.NewRouter(..., onDone)` +- Callback handler gọi `onDone(task)` async sau khi trả response + +--- + +## Pipeline hoàn chỉnh 🎉 + +``` +IMAP IDLE → RECEIVED → AI_PROCESSING → (OpenClaw callback) → CALLBACK_DONE → SMTP reply → COMPLETED + ↓ ↓ ↓ + IGNORED FAILED FAILED +``` + +Tất cả transitions đã được implement: +- IMAP → RECEIVED ✅ +- RECEIVED → AI_PROCESSING ✅ +- AI_PROCESSING → CALLBACK_DONE ✅ +- CALLBACK_DONE → SMTP_RETRYING → COMPLETED/FAILED ✅ +- Anti-loop/Whitelist → IGNORED ✅ +- Retry exhausted → FAILED ✅ + +## Files thay đổi + +- **Mới:** `internal/mail/smtp.go` +- **Cập nhật:** `cmd/bridge/main.go` (wire SMTP sender) + +## Tiếp theo + +1. **Integration tests** — 6 scenarios từ requirements section 10 diff --git a/checkpoints/006-tests.md b/checkpoints/006-tests.md new file mode 100644 index 0000000..5057846 --- /dev/null +++ b/checkpoints/006-tests.md @@ -0,0 +1,44 @@ +# Checkpoint 006 — Tests + +**Ngày:** 2026-04-24 +**Trạng thái:** ✅ Hoàn thành +**Verify:** `go test ./... -v` — 10/10 PASS + +--- + +## Test matrix coverage + +| # | Scenario | Test | Status | +|---|----------|------|--------| +| 1 | Callback hợp lệ → CALLBACK_DONE | `TestCallback_ValidToken_UpdatesTask` | ✅ PASS | +| 2 | OpenClaw fail 2 lần, lần 3 OK → attempt=3 | `TestDispatch_RetryThenSuccess` | ✅ PASS | +| 2' | OpenClaw fail hết → FAILED + last_error | `TestDispatch_AllRetriesFailed` | ✅ PASS | +| 3 | SMTP fail hết retry → FAILED | *(cần SMTP mock — chưa test)* | ⏳ | +| 4 | Callback sai token → 401, state không đổi | `TestCallback_WrongToken_Returns401` | ✅ PASS | +| 5 | Duplicate message_id → không tạo task mới | `TestDuplicateMessageID` | ✅ PASS | +| 6 | Anti-loop / whitelist → IGNORED | *(logic test trong IMAP — chưa test)* | ⏳ | + +**Bonus tests:** +- `TestCallback_AlreadyCompleted_Idempotent` — idempotent callback ✅ +- `TestFindByMessageID_NotFound` — empty result handling ✅ +- `TestThreadHistory` — chỉ trả COMPLETED tasks ✅ +- `TestHealthz`, `TestReadyz` — health endpoints ✅ + +## Kỹ thuật test + +- **Test database:** `database.NewTestDB(t)` tạo SQLite tạm trong `t.TempDir()` +- **HTTP mock:** `httptest.NewServer` cho OpenClaw API +- **HTTP test:** `httptest.NewServer(router)` cho callback handler +- **Retry timing:** Tests thật sự chờ backoff (1s+5s = ~6s mỗi retry test) + +## Files mới + +- `internal/database/testhelper.go` — TestDB helper +- `internal/database/repository_test.go` — repository tests +- `internal/api/router_test.go` — callback + health tests +- `internal/ai_client/client_test.go` — dispatch retry tests + +## Chưa test (cần mock phức tạp hơn) + +- SMTP retry logic (#3) — cần mock `net/smtp.SendMail` +- IMAP anti-loop/whitelist (#6) — cần refactor `processMessage` để testable diff --git a/cmd/bridge/main.go b/cmd/bridge/main.go new file mode 100644 index 0000000..6e661e1 --- /dev/null +++ b/cmd/bridge/main.go @@ -0,0 +1,77 @@ +package main + +import ( + "context" + "log/slog" + "os" + "os/signal" + "sync" + "syscall" + + "thuanle.me/claw-email-bridge/internal/ai_client" + "thuanle.me/claw-email-bridge/internal/api" + "thuanle.me/claw-email-bridge/internal/config" + "thuanle.me/claw-email-bridge/internal/database" + "thuanle.me/claw-email-bridge/internal/logging" + "thuanle.me/claw-email-bridge/internal/mail" +) + +func main() { + logging.Setup() + + cfg, err := config.Load() + if err != nil { + slog.Error("failed to load config", "error", err) + os.Exit(1) + } + slog.Info("config loaded successfully") + + db, err := database.Open("data/database.db") + if err != nil { + slog.Error("failed to open database", "error", err) + os.Exit(1) + } + slog.Info("database ready", "path", "data/database.db") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Graceful shutdown on SIGINT/SIGTERM. + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) + go func() { + sig := <-sigCh + slog.Info("received signal, shutting down", "signal", sig) + cancel() + }() + + var wg sync.WaitGroup + + // Start IMAP watcher with OpenClaw dispatch. + dispatcher := ai_client.NewDispatcher(cfg, db) + watcher := mail.NewIMAPWatcher(cfg, db) + watcher.OnReceived = func(task *database.Task) { + go dispatcher.Dispatch(task) + } + wg.Add(1) + go func() { + defer wg.Done() + watcher.Run(ctx) + }() + + // Start HTTP server with SMTP egress on callback. + smtpSender := mail.NewSMTPSender(cfg, db) + router := api.NewRouter(db, cfg.BridgeCallbackToken, smtpSender.SendReplyFunc()) + wg.Add(1) + go func() { + defer wg.Done() + slog.Info("starting HTTP server", "addr", cfg.ListenAddr) + if err := router.Run(cfg.ListenAddr); err != nil { + slog.Error("server exited with error", "error", err) + cancel() + } + }() + + wg.Wait() + slog.Info("bridge stopped") +} diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..7dbc3ea --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,15 @@ +services: + bridge: + build: . + container_name: claw-email-bridge + restart: unless-stopped + env_file: + - .env + ports: + - "8080:8080" + volumes: + - bridge-data:/app/data + +volumes: + bridge-data: + driver: local diff --git a/docs/specs/2026-04-24-email-openclaw-bridge-design.md b/docs/specs/2026-04-24-email-openclaw-bridge-design.md new file mode 100644 index 0000000..f881c8e --- /dev/null +++ b/docs/specs/2026-04-24-email-openclaw-bridge-design.md @@ -0,0 +1,91 @@ +# Email-OpenClaw Bridge v1 Design (KISS) + +## 1) Scope & Non-goals + +### Scope v1 +- IMAP ingest -> OpenClaw dispatch -> callback -> SMTP reply. +- Stateful processing with SQLite. +- Thread handling via `thread_id`. +- Idempotency by `message_id`. +- Health endpoints: + - `GET /healthz` (liveness) + - `GET /readyz` (readiness) +- Retry mechanism: + - OpenClaw dispatch retries: 3 attempts with backoff 1s, 5s, 15s. + - SMTP send retries: 3 attempts with backoff 1s, 5s, 15s. + +### Non-goals v1 +- No rate limiting. +- No DLQ / reprocess UI. +- No advanced metrics stack. + +## 2) State Machine & Schema + +### Task states +- `IGNORED` +- `RECEIVED` +- `AI_PROCESSING` +- `CALLBACK_DONE` +- `SMTP_RETRYING` +- `COMPLETED` +- `FAILED` + +### Retry rules +- OpenClaw dispatch failure: retry 3 times (1s, 5s, 15s). Exhausted retries -> `FAILED`. +- SMTP send failure: transition to `SMTP_RETRYING`, retry 3 times (1s, 5s, 15s). Exhausted retries -> `FAILED`. +- Persist final failure reason in `last_error`. + +### SQLite schema updates (`tasks`) +- `attempt_openclaw INTEGER DEFAULT 0` +- `attempt_smtp INTEGER DEFAULT 0` +- `last_error TEXT NULL` +- `next_retry_at DATETIME NULL` (used only if retry scheduling is asynchronous) + +## 3) API Contracts & Security + +### OpenClaw callback contract +Required fields: +- `metadata.task_uuid` +- `result` (text) +- `status` + +### Webhook authentication (KISS) +- Bridge requires header `X-Bridge-Token` on callback requests. +- Expected secret is configured in `.env` (e.g., `BRIDGE_CALLBACK_TOKEN`). +- Missing/invalid token -> `401 Unauthorized`. + +### Callback idempotency +- If task already in `COMPLETED`, callback handler returns `200` and skips updates. + +### Health endpoints behavior +- `GET /healthz` -> `200 {"status":"ok"}` when process is alive. +- `GET /readyz` checks: + - DB writable + - required config present (IMAP/SMTP/OpenClaw/token) + - Returns `503` on any failed check. + +## 4) Acceptance Criteria + +1. Valid incoming email eventually reaches `COMPLETED` and reply preserves email thread headers. +2. OpenClaw/SMTP transient failures retry with exact backoff sequence 1s, 5s, 15s. +3. Exhausted retries result in `FAILED` with `last_error` populated. +4. Anti-loop, whitelist, and idempotency behavior work as defined. + +## 5) Test Matrix (v1) + +1. Happy path: new mail -> valid callback -> SMTP success -> `COMPLETED`. +2. OpenClaw fails twice then succeeds -> `COMPLETED`, `attempt_openclaw=3`. +3. SMTP fails all retries -> `FAILED`, `last_error` present. +4. Callback with invalid token -> `401`, no task state change. +5. Duplicate `message_id` -> no new task created. +6. System sender or non-whitelisted sender -> `IGNORED`. + +## 6) Logging (minimum) + +Structured logs must include: +- `task_uuid` +- `thread_id` +- `message_id` +- `status` +- `attempt` +- `error` (when present) diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..1bde937 --- /dev/null +++ b/go.mod @@ -0,0 +1,55 @@ +module thuanle.me/claw-email-bridge + +go 1.26.2 + +require ( + github.com/emersion/go-imap/v2 v2.0.0-beta.8 + github.com/emersion/go-message v0.18.2 + github.com/gin-gonic/gin v1.12.0 + github.com/glebarez/sqlite v1.11.0 + github.com/google/uuid v1.3.0 + github.com/joho/godotenv v1.5.1 + gorm.io/gorm v1.31.1 +) + +require ( + github.com/bytedance/gopkg v0.1.3 // indirect + github.com/bytedance/sonic v1.15.0 // indirect + github.com/bytedance/sonic/loader v0.5.0 // indirect + github.com/cloudwego/base64x v0.1.6 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 // indirect + github.com/gabriel-vasile/mimetype v1.4.12 // indirect + github.com/gin-contrib/sse v1.1.0 // indirect + github.com/glebarez/go-sqlite v1.21.2 // indirect + github.com/go-playground/locales v0.14.1 // indirect + github.com/go-playground/universal-translator v0.18.1 // indirect + github.com/go-playground/validator/v10 v10.30.1 // indirect + github.com/goccy/go-json v0.10.5 // indirect + github.com/goccy/go-yaml v1.19.2 // indirect + github.com/jinzhu/inflection v1.0.0 // indirect + github.com/jinzhu/now v1.1.5 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/klauspost/cpuid/v2 v2.3.0 // indirect + github.com/leodido/go-urn v1.4.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/pelletier/go-toml/v2 v2.2.4 // indirect + github.com/quic-go/qpack v0.6.0 // indirect + github.com/quic-go/quic-go v0.59.0 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + github.com/twitchyliquid64/golang-asm v0.15.1 // indirect + github.com/ugorji/go/codec v1.3.1 // indirect + go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect + golang.org/x/arch v0.22.0 // indirect + golang.org/x/crypto v0.48.0 // indirect + golang.org/x/net v0.51.0 // indirect + golang.org/x/sys v0.41.0 // indirect + golang.org/x/text v0.34.0 // indirect + google.golang.org/protobuf v1.36.10 // indirect + modernc.org/libc v1.22.5 // indirect + modernc.org/mathutil v1.5.0 // indirect + modernc.org/memory v1.5.0 // indirect + modernc.org/sqlite v1.23.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..dceaa97 --- /dev/null +++ b/go.sum @@ -0,0 +1,155 @@ +github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= +github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM= +github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE= +github.com/bytedance/sonic v1.15.0/go.mod h1:tFkWrPz0/CUCLEF4ri4UkHekCIcdnkqXw9VduqpJh0k= +github.com/bytedance/sonic/loader v0.5.0 h1:gXH3KVnatgY7loH5/TkeVyXPfESoqSBSBEiDd5VjlgE= +github.com/bytedance/sonic/loader v0.5.0/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo= +github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M= +github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/emersion/go-imap/v2 v2.0.0-beta.8 h1:5IXZK1E33DyeP526320J3RS7eFlCYGFgtbrfapqDPug= +github.com/emersion/go-imap/v2 v2.0.0-beta.8/go.mod h1:dhoFe2Q0PwLrMD7oZw8ODuaD0vLYPe5uj2wcOMnvh48= +github.com/emersion/go-message v0.18.2 h1:rl55SQdjd9oJcIoQNhubD2Acs1E6IzlZISRTK7x/Lpg= +github.com/emersion/go-message v0.18.2/go.mod h1:XpJyL70LwRvq2a8rVbHXikPgKj8+aI0kGdHlg16ibYA= +github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 h1:oP4q0fw+fOSWn3DfFi4EXdT+B+gTtzx8GC9xsc26Znk= +github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6/go.mod h1:iL2twTeMvZnrg54ZoPDNfJaJaqy0xIQFuBdrLsmspwQ= +github.com/gabriel-vasile/mimetype v1.4.12 h1:e9hWvmLYvtp846tLHam2o++qitpguFiYCKbn0w9jyqw= +github.com/gabriel-vasile/mimetype v1.4.12/go.mod h1:d+9Oxyo1wTzWdyVUPMmXFvp4F9tea18J8ufA774AB3s= +github.com/gin-contrib/sse v1.1.0 h1:n0w2GMuUpWDVp7qSpvze6fAu9iRxJY4Hmj6AmBOU05w= +github.com/gin-contrib/sse v1.1.0/go.mod h1:hxRZ5gVpWMT7Z0B0gSNYqqsSCNIJMjzvm6fqCz9vjwM= +github.com/gin-gonic/gin v1.12.0 h1:b3YAbrZtnf8N//yjKeU2+MQsh2mY5htkZidOM7O0wG8= +github.com/gin-gonic/gin v1.12.0/go.mod h1:VxccKfsSllpKshkBWgVgRniFFAzFb9csfngsqANjnLc= +github.com/glebarez/go-sqlite v1.21.2 h1:3a6LFC4sKahUunAmynQKLZceZCOzUthkRkEAl9gAXWo= +github.com/glebarez/go-sqlite v1.21.2/go.mod h1:sfxdZyhQjTM2Wry3gVYWaW072Ri1WMdWJi0k6+3382k= +github.com/glebarez/sqlite v1.11.0 h1:wSG0irqzP6VurnMEpFGer5Li19RpIRi2qvQz++w0GMw= +github.com/glebarez/sqlite v1.11.0/go.mod h1:h8/o8j5wiAsqSPoWELDUdJXhjAhsVliSn7bWZjOhrgQ= +github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= +github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= +github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA= +github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY= +github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY= +github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY= +github.com/go-playground/validator/v10 v10.30.1 h1:f3zDSN/zOma+w6+1Wswgd9fLkdwy06ntQJp0BBvFG0w= +github.com/go-playground/validator/v10 v10.30.1/go.mod h1:oSuBIQzuJxL//3MelwSLD5hc2Tu889bF0Idm9Dg26cM= +github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4= +github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M= +github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM= +github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/pprof v0.0.0-20221118152302-e6195bd50e26 h1:Xim43kblpZXfIBQsbuBVKCudVG457BR2GZFIz3uw3hQ= +github.com/google/pprof v0.0.0-20221118152302-e6195bd50e26/go.mod h1:dDKJzRmX4S37WGHujM7tX//fmj1uioxKzKxz3lo4HJo= +github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= +github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= +github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= +github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= +github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= +github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= +github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= +github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= +github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= +github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= +github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= +github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= +github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= +github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= +github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY= +github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE= +go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= +go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= +go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU= +golang.org/x/arch v0.22.0 h1:c/Zle32i5ttqRXjdLyyHZESLD/bB90DCU1g9l/0YBDI= +golang.org/x/arch v0.22.0/go.mod h1:dNHoOeKiyja7GTvF9NJS1l3Z2yntpQNzgrjh1cU103A= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= +golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= +golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= +golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= +golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= +golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= +golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= +google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gorm.io/gorm v1.31.1 h1:7CA8FTFz/gRfgqgpeKIBcervUn3xSyPUmr6B2WXJ7kg= +gorm.io/gorm v1.31.1/go.mod h1:XyQVbO2k6YkOis7C2437jSit3SsDK72s7n7rsSHd+Gs= +modernc.org/libc v1.22.5 h1:91BNch/e5B0uPbJFgqbxXuOnxBQjlS//icfQEGmvyjE= +modernc.org/libc v1.22.5/go.mod h1:jj+Z7dTNX8fBScMVNRAYZ/jF91K8fdT2hYMThc3YjBY= +modernc.org/mathutil v1.5.0 h1:rV0Ko/6SfM+8G+yKiyI830l3Wuz1zRutdslNoQ0kfiQ= +modernc.org/mathutil v1.5.0/go.mod h1:mZW8CKdRPY1v87qxC/wUdX5O1qDzXMP5TH3wjfpga6E= +modernc.org/memory v1.5.0 h1:N+/8c5rE6EqugZwHii4IFsaJ7MUhoWX07J5tC/iI5Ds= +modernc.org/memory v1.5.0/go.mod h1:PkUhL0Mugw21sHPeskwZW4D6VscE/GQJOnIpCnW6pSU= +modernc.org/sqlite v1.23.1 h1:nrSBg4aRQQwq59JpvGEQ15tNxoO5pX/kUjcRNwSAGQM= +modernc.org/sqlite v1.23.1/go.mod h1:OrDj17Mggn6MhE+iPbBNf7RGKODDE9NFT0f3EwDzJqk= diff --git a/internal/ai_client/client.go b/internal/ai_client/client.go new file mode 100644 index 0000000..95c7f81 --- /dev/null +++ b/internal/ai_client/client.go @@ -0,0 +1,188 @@ +package ai_client + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "time" + + "thuanle.me/claw-email-bridge/internal/config" + "thuanle.me/claw-email-bridge/internal/database" + "thuanle.me/claw-email-bridge/internal/logging" + "gorm.io/gorm" +) + +// Retry backoff schedule: 1s, 5s, 15s. +var retryBackoffs = []time.Duration{1 * time.Second, 5 * time.Second, 15 * time.Second} + +// openClawRequest is the payload sent to OpenClaw API. +type openClawRequest struct { + Input string `json:"input"` + SessionID string `json:"session_id"` + History []historyEntry `json:"history,omitempty"` + CallbackURL string `json:"callback_url"` + Metadata map[string]string `json:"metadata"` +} + +type historyEntry struct { + Role string `json:"role"` + Content string `json:"content"` +} + +// Dispatcher sends tasks to OpenClaw and handles retry logic. +type Dispatcher struct { + cfg *config.Config + db *gorm.DB + httpClient *http.Client +} + +// NewDispatcher creates a dispatcher with a direct HTTP client (no system proxy). +func NewDispatcher(cfg *config.Config, db *gorm.DB) *Dispatcher { + return &Dispatcher{ + cfg: cfg, + db: db, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + Transport: &http.Transport{ + Proxy: nil, // Direct connection, no proxy inheritance. + }, + }, + } +} + +// Dispatch sends a task to OpenClaw with retry. +// This should be called asynchronously (go dispatcher.Dispatch(task)). +func (d *Dispatcher) Dispatch(task *database.Task) { + log := logging.TaskLogger(task.TaskUUID, task.ThreadID, task.MessageID) + + // Build history from completed tasks in the same thread. + history, err := d.buildHistory(task.ThreadID) + if err != nil { + log.Error("openclaw: failed to build history", "error", err) + // Continue without history — non-fatal. + } + + // Build callback URL. + callbackURL := fmt.Sprintf("http://%s/callback", d.cfg.ListenAddr) + + payload := openClawRequest{ + Input: task.BodyPlain, + SessionID: task.ThreadID, + History: history, + CallbackURL: callbackURL, + Metadata: map[string]string{"task_uuid": task.TaskUUID}, + } + + // Update status to AI_PROCESSING before first attempt. + task.Status = database.StatusAIProcessing + if err := database.UpdateTask(d.db, task); err != nil { + log.Error("openclaw: failed to update status to AI_PROCESSING", "error", err) + return + } + + // Retry loop: up to 3 attempts. + for attempt := 1; attempt <= len(retryBackoffs); attempt++ { + task.AttemptOpenClaw = attempt + + log.Info("openclaw: dispatching", + "attempt", attempt, + "status", task.Status, + ) + + err := d.callAPI(payload) + if err == nil { + // Success — task stays AI_PROCESSING, waiting for callback. + log.Info("openclaw: dispatch successful", "attempt", attempt) + if err := database.UpdateTask(d.db, task); err != nil { + log.Error("openclaw: failed to save attempt count", "error", err) + } + return + } + + log.Warn("openclaw: dispatch failed", + "attempt", attempt, + "error", err, + ) + + // Last attempt — mark as FAILED. + if attempt == len(retryBackoffs) { + errMsg := err.Error() + task.Status = database.StatusFailed + task.LastError = &errMsg + if err := database.UpdateTask(d.db, task); err != nil { + log.Error("openclaw: failed to save FAILED status", "error", err) + } + log.Error("openclaw: all retries exhausted", + "attempt", attempt, + "status", database.StatusFailed, + "error", errMsg, + ) + return + } + + // Wait before retry. + backoff := retryBackoffs[attempt-1] + log.Info("openclaw: retrying after backoff", "backoff", backoff) + time.Sleep(backoff) + } +} + +// callAPI makes a single HTTP POST to OpenClaw. +func (d *Dispatcher) callAPI(payload openClawRequest) error { + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("marshal payload: %w", err) + } + + req, err := http.NewRequest(http.MethodPost, d.cfg.OpenClawURL, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + if d.cfg.OpenClawAPIKey != "" { + req.Header.Set("Authorization", "Bearer "+d.cfg.OpenClawAPIKey) + } + + resp, err := d.httpClient.Do(req) + if err != nil { + return fmt.Errorf("http request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return nil + } + + // Read response body for error details (limit to 1KB). + respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) + return fmt.Errorf("openclaw returned %d: %s", resp.StatusCode, string(respBody)) +} + +// buildHistory fetches completed tasks in the thread and formats them as history entries. +func (d *Dispatcher) buildHistory(threadID string) ([]historyEntry, error) { + tasks, err := database.ThreadHistory(d.db, threadID) + if err != nil { + return nil, err + } + + var history []historyEntry + for _, t := range tasks { + // User message. + history = append(history, historyEntry{ + Role: "user", + Content: t.BodyPlain, + }) + // AI response. + if t.AIResponse != "" { + history = append(history, historyEntry{ + Role: "assistant", + Content: t.AIResponse, + }) + } + } + + return history, nil +} diff --git a/internal/ai_client/client_test.go b/internal/ai_client/client_test.go new file mode 100644 index 0000000..77b8643 --- /dev/null +++ b/internal/ai_client/client_test.go @@ -0,0 +1,114 @@ +package ai_client_test + +import ( + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "thuanle.me/claw-email-bridge/internal/ai_client" + "thuanle.me/claw-email-bridge/internal/config" + "thuanle.me/claw-email-bridge/internal/database" +) + +// Test matrix #2: OpenClaw fail 2 lần, lần 3 thành công → COMPLETED, attempt_openclaw = 3. +func TestDispatch_RetryThenSuccess(t *testing.T) { + tdb := database.NewTestDB(t) + + var callCount atomic.Int32 + mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := callCount.Add(1) + if n <= 2 { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte("temporary error")) + return + } + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"status":"ok"}`)) + })) + defer mockServer.Close() + + cfg := &config.Config{ + OpenClawURL: mockServer.URL, + ListenAddr: ":9999", + } + + task := &database.Task{ + TaskUUID: "uuid-retry-test", + ThreadID: "thread-1", + MessageID: "msg-retry@test.com", + Sender: "user@test.com", + Subject: "Retry test", + BodyPlain: "Hello", + Status: database.StatusReceived, + } + if err := database.CreateTask(tdb.DB, task); err != nil { + t.Fatalf("create task: %v", err) + } + + dispatcher := ai_client.NewDispatcher(cfg, tdb.DB) + dispatcher.Dispatch(task) // Synchronous call for testing. + + // Verify 3 attempts were made. + if callCount.Load() != 3 { + t.Errorf("expected 3 API calls, got %d", callCount.Load()) + } + + // Reload task from DB. + updated, err := database.FindByTaskUUID(tdb.DB, "uuid-retry-test") + if err != nil { + t.Fatalf("find task: %v", err) + } + if updated.Status != database.StatusAIProcessing { + t.Errorf("expected AI_PROCESSING (waiting for callback), got %s", updated.Status) + } + if updated.AttemptOpenClaw != 3 { + t.Errorf("expected attempt_openclaw = 3, got %d", updated.AttemptOpenClaw) + } +} + +// OpenClaw fail all 3 retries → FAILED with last_error. +func TestDispatch_AllRetriesFailed(t *testing.T) { + tdb := database.NewTestDB(t) + + mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte("server down")) + })) + defer mockServer.Close() + + cfg := &config.Config{ + OpenClawURL: mockServer.URL, + ListenAddr: ":9999", + } + + task := &database.Task{ + TaskUUID: "uuid-fail-all", + ThreadID: "thread-1", + MessageID: "msg-fail@test.com", + Sender: "user@test.com", + Subject: "Fail test", + BodyPlain: "Hello", + Status: database.StatusReceived, + } + if err := database.CreateTask(tdb.DB, task); err != nil { + t.Fatalf("create task: %v", err) + } + + dispatcher := ai_client.NewDispatcher(cfg, tdb.DB) + dispatcher.Dispatch(task) + + updated, err := database.FindByTaskUUID(tdb.DB, "uuid-fail-all") + if err != nil { + t.Fatalf("find task: %v", err) + } + if updated.Status != database.StatusFailed { + t.Errorf("expected FAILED, got %s", updated.Status) + } + if updated.LastError == nil { + t.Error("expected last_error to be set") + } + if updated.AttemptOpenClaw != 3 { + t.Errorf("expected attempt_openclaw = 3, got %d", updated.AttemptOpenClaw) + } +} diff --git a/internal/api/router.go b/internal/api/router.go new file mode 100644 index 0000000..b8a3d1e --- /dev/null +++ b/internal/api/router.go @@ -0,0 +1,137 @@ +package api + +import ( + "log/slog" + "net/http" + + "github.com/gin-gonic/gin" + "thuanle.me/claw-email-bridge/internal/database" + "thuanle.me/claw-email-bridge/internal/logging" + "gorm.io/gorm" +) + +// CallbackRequest is the payload received from OpenClaw. +type CallbackRequest struct { + Metadata struct { + TaskUUID string `json:"task_uuid" binding:"required"` + } `json:"metadata" binding:"required"` + Result string `json:"result" binding:"required"` + Status string `json:"status"` +} + +// OnCallbackDone is called after a task transitions to CALLBACK_DONE. +// This will be wired to SMTP egress. +type OnCallbackDone func(task *database.Task) + +// NewRouter creates the Gin engine with health and callback routes. +func NewRouter(db *gorm.DB, callbackToken string, onDone OnCallbackDone) *gin.Engine { + gin.SetMode(gin.ReleaseMode) + r := gin.New() + r.Use(gin.Recovery()) + + r.GET("/healthz", func(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{"status": "ok"}) + }) + + r.GET("/readyz", readyzHandler(db, callbackToken)) + r.POST("/callback", tokenAuthMiddleware(callbackToken), callbackHandler(db, onDone)) + + return r +} + +// tokenAuthMiddleware validates X-Bridge-Token header. +func tokenAuthMiddleware(expected string) gin.HandlerFunc { + return func(c *gin.Context) { + token := c.GetHeader("X-Bridge-Token") + if token == "" || token != expected { + slog.Warn("callback: unauthorized request", + "remote_addr", c.ClientIP(), + "has_token", token != "", + ) + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{ + "error": "unauthorized", + }) + return + } + c.Next() + } +} + +// callbackHandler processes the callback from OpenClaw. +func callbackHandler(db *gorm.DB, onDone OnCallbackDone) gin.HandlerFunc { + return func(c *gin.Context) { + var req CallbackRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload: " + err.Error()}) + return + } + + taskUUID := req.Metadata.TaskUUID + log := logging.TaskLogger(taskUUID, "", "") + + // Find task. + task, err := database.FindByTaskUUID(db, taskUUID) + if err != nil { + log.Error("callback: db lookup failed", "error", err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) + return + } + if task == nil { + log.Warn("callback: task not found") + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + + log = logging.TaskLogger(task.TaskUUID, task.ThreadID, task.MessageID) + + // Idempotent: if already COMPLETED, return 200 without changes. + if task.Status == database.StatusCompleted { + log.Info("callback: task already completed, idempotent response") + c.JSON(http.StatusOK, gin.H{"status": "already_completed"}) + return + } + + // Update task with AI response. + task.AIResponse = req.Result + task.Status = database.StatusCallbackDone + if err := database.UpdateTask(db, task); err != nil { + log.Error("callback: failed to update task", "error", err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) + return + } + + log.Info("callback: task updated", + "status", database.StatusCallbackDone, + ) + + c.JSON(http.StatusOK, gin.H{"status": "ok"}) + + // Trigger SMTP egress (async, after response is sent). + if onDone != nil { + go onDone(task) + } + } +} + +// readyzHandler checks DB writable + required config present. +func readyzHandler(db *gorm.DB, callbackToken string) gin.HandlerFunc { + return func(c *gin.Context) { + if err := database.Healthy(db); err != nil { + c.JSON(http.StatusServiceUnavailable, gin.H{ + "status": "error", + "detail": "database not writable", + }) + return + } + + if callbackToken == "" { + c.JSON(http.StatusServiceUnavailable, gin.H{ + "status": "error", + "detail": "missing BRIDGE_CALLBACK_TOKEN", + }) + return + } + + c.JSON(http.StatusOK, gin.H{"status": "ok"}) + } +} diff --git a/internal/api/router_test.go b/internal/api/router_test.go new file mode 100644 index 0000000..f67cf31 --- /dev/null +++ b/internal/api/router_test.go @@ -0,0 +1,198 @@ +package api_test + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "thuanle.me/claw-email-bridge/internal/api" + "thuanle.me/claw-email-bridge/internal/database" +) + +const testToken = "test-secret-token" + +func setupTestRouter(t *testing.T) (*database.TestDB, *httptest.Server) { + t.Helper() + tdb := database.NewTestDB(t) + router := api.NewRouter(tdb.DB, testToken, nil) + srv := httptest.NewServer(router) + t.Cleanup(func() { srv.Close() }) + return tdb, srv +} + +// Test matrix #4: Callback sai token → 401, state không đổi. +func TestCallback_WrongToken_Returns401(t *testing.T) { + tdb, srv := setupTestRouter(t) + + // Create a task in AI_PROCESSING state. + task := &database.Task{ + TaskUUID: "uuid-wrong-token", + ThreadID: "thread-1", + MessageID: "msg-1", + Sender: "user@test.com", + Subject: "Test", + Status: database.StatusAIProcessing, + } + if err := database.CreateTask(tdb.DB, task); err != nil { + t.Fatalf("create task: %v", err) + } + + payload := map[string]any{ + "metadata": map[string]string{"task_uuid": "uuid-wrong-token"}, + "result": "AI response", + "status": "ok", + } + body, _ := json.Marshal(payload) + + // No token. + resp, err := http.Post(srv.URL+"/callback", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatalf("request failed: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusUnauthorized { + t.Errorf("expected 401, got %d", resp.StatusCode) + } + + // Wrong token. + req, _ := http.NewRequest("POST", srv.URL+"/callback", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Bridge-Token", "wrong-token") + resp2, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("request failed: %v", err) + } + defer resp2.Body.Close() + if resp2.StatusCode != http.StatusUnauthorized { + t.Errorf("expected 401, got %d", resp2.StatusCode) + } + + // Verify state unchanged. + updated, err := database.FindByTaskUUID(tdb.DB, "uuid-wrong-token") + if err != nil { + t.Fatalf("find task: %v", err) + } + if updated.Status != database.StatusAIProcessing { + t.Errorf("expected status AI_PROCESSING, got %s", updated.Status) + } + if updated.AIResponse != "" { + t.Errorf("expected empty ai_response, got %q", updated.AIResponse) + } +} + +// Test matrix #1 (partial): callback hợp lệ → CALLBACK_DONE. +func TestCallback_ValidToken_UpdatesTask(t *testing.T) { + tdb, srv := setupTestRouter(t) + + task := &database.Task{ + TaskUUID: "uuid-valid", + ThreadID: "thread-1", + MessageID: "msg-1", + Sender: "user@test.com", + Subject: "Test", + Status: database.StatusAIProcessing, + } + if err := database.CreateTask(tdb.DB, task); err != nil { + t.Fatalf("create task: %v", err) + } + + payload := map[string]any{ + "metadata": map[string]string{"task_uuid": "uuid-valid"}, + "result": "Hello from AI", + "status": "ok", + } + body, _ := json.Marshal(payload) + + req, _ := http.NewRequest("POST", srv.URL+"/callback", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Bridge-Token", testToken) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("request failed: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Errorf("expected 200, got %d", resp.StatusCode) + } + + updated, _ := database.FindByTaskUUID(tdb.DB, "uuid-valid") + if updated.Status != database.StatusCallbackDone { + t.Errorf("expected CALLBACK_DONE, got %s", updated.Status) + } + if updated.AIResponse != "Hello from AI" { + t.Errorf("expected 'Hello from AI', got %q", updated.AIResponse) + } +} + +// Idempotent callback: task already COMPLETED → 200, no changes. +func TestCallback_AlreadyCompleted_Idempotent(t *testing.T) { + tdb, srv := setupTestRouter(t) + + task := &database.Task{ + TaskUUID: "uuid-completed", + ThreadID: "thread-1", + MessageID: "msg-1", + Sender: "user@test.com", + Subject: "Test", + Status: database.StatusCompleted, + AIResponse: "Original response", + } + if err := database.CreateTask(tdb.DB, task); err != nil { + t.Fatalf("create task: %v", err) + } + + payload := map[string]any{ + "metadata": map[string]string{"task_uuid": "uuid-completed"}, + "result": "New response that should be ignored", + "status": "ok", + } + body, _ := json.Marshal(payload) + + req, _ := http.NewRequest("POST", srv.URL+"/callback", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Bridge-Token", testToken) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("request failed: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Errorf("expected 200, got %d", resp.StatusCode) + } + + updated, _ := database.FindByTaskUUID(tdb.DB, "uuid-completed") + if updated.AIResponse != "Original response" { + t.Errorf("expected original response preserved, got %q", updated.AIResponse) + } +} + +// Health endpoints. +func TestHealthz(t *testing.T) { + _, srv := setupTestRouter(t) + + resp, err := http.Get(srv.URL + "/healthz") + if err != nil { + t.Fatalf("request failed: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Errorf("expected 200, got %d", resp.StatusCode) + } +} + +func TestReadyz(t *testing.T) { + _, srv := setupTestRouter(t) + + resp, err := http.Get(srv.URL + "/readyz") + if err != nil { + t.Fatalf("request failed: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Errorf("expected 200, got %d", resp.StatusCode) + } +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..6c8197d --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,111 @@ +package config + +import ( + "fmt" + "os" + "strings" + + "github.com/joho/godotenv" +) + +// Config holds all application configuration loaded from environment variables. +type Config struct { + // IMAP settings + IMAPHost string + IMAPPort string + IMAPUser string + IMAPPass string + + // SMTP settings + SMTPHost string + SMTPPort string + SMTPUser string + SMTPPass string + + // OpenClaw settings + OpenClawURL string + OpenClawAPIKey string + + // Bridge settings + BridgeCallbackToken string + SystemEmail string + WhitelistEmails []string + + // Optional + IMAPProxyURL string + + // Server + ListenAddr string +} + +// Load reads .env (if present) and populates Config from environment variables. +// Returns an error if any required variable is missing. +func Load() (*Config, error) { + // Best-effort load .env — ignore error if file doesn't exist. + _ = godotenv.Load() + + cfg := &Config{ + IMAPHost: os.Getenv("IMAP_HOST"), + IMAPPort: os.Getenv("IMAP_PORT"), + IMAPUser: os.Getenv("IMAP_USER"), + IMAPPass: os.Getenv("IMAP_PASS"), + SMTPHost: os.Getenv("SMTP_HOST"), + SMTPPort: os.Getenv("SMTP_PORT"), + SMTPUser: os.Getenv("SMTP_USER"), + SMTPPass: os.Getenv("SMTP_PASS"), + OpenClawURL: os.Getenv("OPENCLAW_URL"), + OpenClawAPIKey: os.Getenv("OPENCLAW_API_KEY"), + BridgeCallbackToken: os.Getenv("BRIDGE_CALLBACK_TOKEN"), + SystemEmail: os.Getenv("SYSTEM_EMAIL"), + IMAPProxyURL: os.Getenv("IMAP_PROXY_URL"), + ListenAddr: os.Getenv("LISTEN_ADDR"), + } + + if raw := os.Getenv("WHITELIST_EMAILS"); raw != "" { + for _, email := range strings.Split(raw, ",") { + if trimmed := strings.TrimSpace(email); trimmed != "" { + cfg.WhitelistEmails = append(cfg.WhitelistEmails, trimmed) + } + } + } + + if cfg.ListenAddr == "" { + cfg.ListenAddr = ":8080" + } + + if err := cfg.validate(); err != nil { + return nil, err + } + + return cfg, nil +} + +// validate checks that all required configuration values are present. +func (c *Config) validate() error { + required := map[string]string{ + "IMAP_HOST": c.IMAPHost, + "IMAP_PORT": c.IMAPPort, + "IMAP_USER": c.IMAPUser, + "IMAP_PASS": c.IMAPPass, + "SMTP_HOST": c.SMTPHost, + "SMTP_PORT": c.SMTPPort, + "SMTP_USER": c.SMTPUser, + "SMTP_PASS": c.SMTPPass, + "OPENCLAW_URL": c.OpenClawURL, + "BRIDGE_CALLBACK_TOKEN": c.BridgeCallbackToken, + "SYSTEM_EMAIL": c.SystemEmail, + } + + var missing []string + for name, val := range required { + if val == "" { + missing = append(missing, name) + } + } + + if len(missing) > 0 { + return fmt.Errorf("missing required config: %s", strings.Join(missing, ", ")) + } + + return nil +} diff --git a/internal/database/database.go b/internal/database/database.go new file mode 100644 index 0000000..78583af --- /dev/null +++ b/internal/database/database.go @@ -0,0 +1,35 @@ +package database + +import ( + "fmt" + + "github.com/glebarez/sqlite" + "gorm.io/gorm" + "gorm.io/gorm/logger" +) + +// Open initializes the SQLite database and runs auto-migration. +func Open(dbPath string) (*gorm.DB, error) { + db, err := gorm.Open(sqlite.Open(dbPath), &gorm.Config{ + Logger: logger.Default.LogMode(logger.Silent), // We use our own structured logger. + }) + if err != nil { + return nil, fmt.Errorf("database open: %w", err) + } + + // Enable WAL mode for better concurrent read performance. + if err := db.Exec("PRAGMA journal_mode=WAL").Error; err != nil { + return nil, fmt.Errorf("database pragma WAL: %w", err) + } + + if err := db.AutoMigrate(&Task{}); err != nil { + return nil, fmt.Errorf("database migrate: %w", err) + } + + return db, nil +} + +// Healthy checks if the database is writable. +func Healthy(db *gorm.DB) error { + return db.Exec("SELECT 1").Error +} diff --git a/internal/database/models.go b/internal/database/models.go new file mode 100644 index 0000000..5fe3db1 --- /dev/null +++ b/internal/database/models.go @@ -0,0 +1,35 @@ +package database + +import ( + "time" +) + +// Task status constants matching the requirement's state machine. +const ( + StatusIgnored = "IGNORED" + StatusReceived = "RECEIVED" + StatusAIProcessing = "AI_PROCESSING" + StatusCallbackDone = "CALLBACK_DONE" + StatusSMTPRetrying = "SMTP_RETRYING" + StatusCompleted = "COMPLETED" + StatusFailed = "FAILED" +) + +// Task represents a single email processing unit tracked through the pipeline. +type Task struct { + ID uint `gorm:"primaryKey;autoIncrement"` + TaskUUID string `gorm:"uniqueIndex;not null"` + ThreadID string `gorm:"index"` + MessageID string `gorm:"index"` + Sender string + Subject string + BodyPlain string + AIResponse string + Status string `gorm:"index;not null"` + AttemptOpenClaw int `gorm:"default:0"` + AttemptSMTP int `gorm:"default:0"` + LastError *string + NextRetryAt *time.Time + CreatedAt time.Time + UpdatedAt time.Time +} diff --git a/internal/database/repository.go b/internal/database/repository.go new file mode 100644 index 0000000..a24a902 --- /dev/null +++ b/internal/database/repository.go @@ -0,0 +1,65 @@ +package database + +import ( + "errors" + + "gorm.io/gorm" +) + +// FindByMessageID returns the task with the given message_id, or nil if not found. +func FindByMessageID(db *gorm.DB, messageID string) (*Task, error) { + var task Task + err := db.Where("message_id = ?", messageID).First(&task).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &task, nil +} + +// FindLatestByThreadID returns the most recent COMPLETED task in the thread. +func FindLatestByThreadID(db *gorm.DB, threadID string) (*Task, error) { + var task Task + err := db.Where("thread_id = ? AND status = ?", threadID, StatusCompleted). + Order("created_at DESC").First(&task).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &task, nil +} + +// FindByTaskUUID returns the task with the given task_uuid. +func FindByTaskUUID(db *gorm.DB, taskUUID string) (*Task, error) { + var task Task + err := db.Where("task_uuid = ?", taskUUID).First(&task).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &task, nil +} + +// ThreadHistory returns all COMPLETED tasks for a thread, ordered chronologically. +func ThreadHistory(db *gorm.DB, threadID string) ([]Task, error) { + var tasks []Task + err := db.Where("thread_id = ? AND status = ?", threadID, StatusCompleted). + Order("created_at ASC").Find(&tasks).Error + return tasks, err +} + +// CreateTask inserts a new task record. +func CreateTask(db *gorm.DB, task *Task) error { + return db.Create(task).Error +} + +// UpdateTask saves changes to an existing task record. +func UpdateTask(db *gorm.DB, task *Task) error { + return db.Save(task).Error +} diff --git a/internal/database/repository_test.go b/internal/database/repository_test.go new file mode 100644 index 0000000..0e2c125 --- /dev/null +++ b/internal/database/repository_test.go @@ -0,0 +1,105 @@ +package database_test + +import ( + "testing" + + "thuanle.me/claw-email-bridge/internal/database" +) + +// Test matrix #5: Duplicate message_id → không tạo task mới. +func TestDuplicateMessageID(t *testing.T) { + tdb := database.NewTestDB(t) + + task1 := &database.Task{ + TaskUUID: "uuid-1", + ThreadID: "thread-1", + MessageID: "msg-duplicate@test.com", + Sender: "user@test.com", + Subject: "First email", + Status: database.StatusReceived, + } + if err := database.CreateTask(tdb.DB, task1); err != nil { + t.Fatalf("create first task: %v", err) + } + + // Lookup should find the existing task. + found, err := database.FindByMessageID(tdb.DB, "msg-duplicate@test.com") + if err != nil { + t.Fatalf("find by message_id: %v", err) + } + if found == nil { + t.Fatal("expected to find existing task") + } + if found.TaskUUID != "uuid-1" { + t.Errorf("expected uuid-1, got %s", found.TaskUUID) + } + + // Attempting to create another task with same task_uuid should fail (unique constraint). + task2 := &database.Task{ + TaskUUID: "uuid-1", // Same UUID — unique constraint violation. + ThreadID: "thread-1", + MessageID: "msg-duplicate@test.com", + Sender: "user@test.com", + Subject: "Duplicate", + Status: database.StatusReceived, + } + err = database.CreateTask(tdb.DB, task2) + if err == nil { + t.Error("expected error creating duplicate task_uuid, got nil") + } +} + +func TestFindByMessageID_NotFound(t *testing.T) { + tdb := database.NewTestDB(t) + + found, err := database.FindByMessageID(tdb.DB, "nonexistent@test.com") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if found != nil { + t.Error("expected nil for nonexistent message_id") + } +} + +func TestThreadHistory(t *testing.T) { + tdb := database.NewTestDB(t) + + // Create two completed tasks in same thread. + for i, uuid := range []string{"uuid-a", "uuid-b"} { + task := &database.Task{ + TaskUUID: uuid, + ThreadID: "thread-history", + MessageID: uuid + "@test.com", + Sender: "user@test.com", + Subject: "Thread test", + BodyPlain: "Message " + uuid, + AIResponse: "Reply " + uuid, + Status: database.StatusCompleted, + } + _ = i + if err := database.CreateTask(tdb.DB, task); err != nil { + t.Fatalf("create task %s: %v", uuid, err) + } + } + + // Create one non-completed task — should not appear in history. + task3 := &database.Task{ + TaskUUID: "uuid-c", + ThreadID: "thread-history", + MessageID: "uuid-c@test.com", + Sender: "user@test.com", + Subject: "Pending", + Status: database.StatusAIProcessing, + } + if err := database.CreateTask(tdb.DB, task3); err != nil { + t.Fatalf("create task: %v", err) + } + + history, err := database.ThreadHistory(tdb.DB, "thread-history") + if err != nil { + t.Fatalf("thread history: %v", err) + } + if len(history) != 2 { + t.Errorf("expected 2 completed tasks in history, got %d", len(history)) + } +} diff --git a/internal/database/testhelper.go b/internal/database/testhelper.go new file mode 100644 index 0000000..f45de49 --- /dev/null +++ b/internal/database/testhelper.go @@ -0,0 +1,35 @@ +package database + +import ( + "os" + "path/filepath" + "testing" + + "gorm.io/gorm" +) + +// TestDB holds a test database instance with automatic cleanup. +type TestDB struct { + DB *gorm.DB + cleanup func() +} + +// NewTestDB creates a temporary SQLite database for testing. +func NewTestDB(t *testing.T) *TestDB { + t.Helper() + + dir := t.TempDir() + dbPath := filepath.Join(dir, "test.db") + + db, err := Open(dbPath) + if err != nil { + t.Fatalf("open test db: %v", err) + } + + return &TestDB{ + DB: db, + cleanup: func() { + os.RemoveAll(dir) + }, + } +} diff --git a/internal/logging/logger.go b/internal/logging/logger.go new file mode 100644 index 0000000..a4256a6 --- /dev/null +++ b/internal/logging/logger.go @@ -0,0 +1,25 @@ +package logging + +import ( + "log/slog" + "os" +) + +// Setup initialises the application-wide structured logger. +// All output goes to stdout as JSON for easy parsing by Docker / log aggregators. +func Setup() { + handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{ + Level: slog.LevelInfo, + }) + slog.SetDefault(slog.New(handler)) +} + +// TaskLogger returns a logger pre-populated with common task fields. +// Usage: logging.TaskLogger(taskUUID, threadID, messageID).Info("message", ...) +func TaskLogger(taskUUID, threadID, messageID string) *slog.Logger { + return slog.With( + "task_uuid", taskUUID, + "thread_id", threadID, + "message_id", messageID, + ) +} diff --git a/internal/mail/imap.go b/internal/mail/imap.go new file mode 100644 index 0000000..c3e7dd5 --- /dev/null +++ b/internal/mail/imap.go @@ -0,0 +1,362 @@ +package mail + +import ( + "context" + "fmt" + "io" + "log/slog" + "strings" + "time" + + gomessage "github.com/emersion/go-message/mail" + + "github.com/emersion/go-imap/v2" + "github.com/emersion/go-imap/v2/imapclient" + "github.com/google/uuid" + "thuanle.me/claw-email-bridge/internal/config" + "thuanle.me/claw-email-bridge/internal/database" + "thuanle.me/claw-email-bridge/internal/logging" + "gorm.io/gorm" +) + +// IMAPWatcher monitors an IMAP mailbox via IDLE and processes new emails. +type IMAPWatcher struct { + cfg *config.Config + db *gorm.DB + client *imapclient.Client + + // onReceived is called after a task is saved as RECEIVED. + // This will be wired to the OpenClaw dispatch in a future step. + OnReceived func(task *database.Task) +} + +// NewIMAPWatcher creates a new IMAP watcher. +func NewIMAPWatcher(cfg *config.Config, db *gorm.DB) *IMAPWatcher { + return &IMAPWatcher{cfg: cfg, db: db} +} + +// Run connects to IMAP, selects INBOX, and enters the IDLE loop. +// It blocks until ctx is cancelled. Reconnects automatically on error. +func (w *IMAPWatcher) Run(ctx context.Context) { + for { + if err := w.connectAndWatch(ctx); err != nil { + slog.Error("imap watcher error, reconnecting in 10s", "error", err) + } + + select { + case <-ctx.Done(): + slog.Info("imap watcher stopped") + return + case <-time.After(10 * time.Second): + // Reconnect after delay. + } + } +} + +// connectAndWatch handles a single IMAP session lifecycle. +func (w *IMAPWatcher) connectAndWatch(ctx context.Context) error { + numMessages := uint32(0) + + options := &imapclient.Options{ + UnilateralDataHandler: &imapclient.UnilateralDataHandler{ + Mailbox: func(data *imapclient.UnilateralDataMailbox) { + if data.NumMessages != nil && *data.NumMessages > numMessages { + slog.Info("imap: new message notification", "num_messages", *data.NumMessages) + numMessages = *data.NumMessages + } + }, + }, + } + + addr := fmt.Sprintf("%s:%s", w.cfg.IMAPHost, w.cfg.IMAPPort) + slog.Info("imap: connecting", "addr", addr) + + c, err := imapclient.DialTLS(addr, options) + if err != nil { + return fmt.Errorf("dial TLS: %w", err) + } + w.client = c + defer func() { + _ = c.Close() + w.client = nil + }() + + if err := c.Login(w.cfg.IMAPUser, w.cfg.IMAPPass).Wait(); err != nil { + return fmt.Errorf("login: %w", err) + } + slog.Info("imap: logged in", "user", w.cfg.IMAPUser) + + selectedMbox, err := c.Select("INBOX", nil).Wait() + if err != nil { + return fmt.Errorf("select INBOX: %w", err) + } + numMessages = selectedMbox.NumMessages + slog.Info("imap: INBOX selected", "messages", numMessages) + + // Fetch any unseen messages on startup. + if err := w.fetchUnseen(c); err != nil { + slog.Error("imap: initial fetch unseen failed", "error", err) + } + + // IDLE loop: wait for new messages, then fetch unseen. + for { + select { + case <-ctx.Done(): + return nil + default: + } + + idleCmd, err := c.Idle() + if err != nil { + return fmt.Errorf("idle: %w", err) + } + + done := make(chan error, 1) + go func() { + done <- idleCmd.Wait() + }() + + // Re-check every 28 minutes (RFC recommends <29 min IDLE timeout). + idleTimer := time.NewTimer(28 * time.Minute) + + select { + case <-ctx.Done(): + _ = idleCmd.Close() + idleTimer.Stop() + return nil + case err := <-done: + idleTimer.Stop() + if err != nil { + return fmt.Errorf("idle wait: %w", err) + } + case <-idleTimer.C: + // Stop IDLE and restart it. + if err := idleCmd.Close(); err != nil { + return fmt.Errorf("idle close: %w", err) + } + <-done + } + + // After exiting IDLE, fetch unseen messages. + if err := w.fetchUnseen(c); err != nil { + slog.Error("imap: fetch unseen failed", "error", err) + } + } +} + +// fetchUnseen searches for UNSEEN messages and processes each one. +func (w *IMAPWatcher) fetchUnseen(c *imapclient.Client) error { + criteria := &imap.SearchCriteria{ + NotFlag: []imap.Flag{imap.FlagSeen}, + } + searchData, err := c.UIDSearch(criteria, nil).Wait() + if err != nil { + return fmt.Errorf("uid search: %w", err) + } + + uids := searchData.AllUIDs() + if len(uids) == 0 { + return nil + } + slog.Info("imap: unseen messages found", "count", len(uids)) + + uidSet := imap.UIDSet{} + for _, uid := range uids { + uidSet.AddNum(uid) + } + + bodySection := &imap.FetchItemBodySection{} + fetchOptions := &imap.FetchOptions{ + Envelope: true, + UID: true, + BodySection: []*imap.FetchItemBodySection{bodySection}, + } + + fetchCmd := c.Fetch(uidSet, fetchOptions) + defer fetchCmd.Close() + + for { + msg := fetchCmd.Next() + if msg == nil { + break + } + + buf, err := msg.Collect() + if err != nil { + slog.Error("imap: collect message failed", "error", err) + continue + } + + w.processMessage(c, buf, bodySection) + } + + if err := fetchCmd.Close(); err != nil { + return fmt.Errorf("fetch close: %w", err) + } + + return nil +} + +// processMessage handles a single fetched email message. +func (w *IMAPWatcher) processMessage(c *imapclient.Client, buf *imapclient.FetchMessageBuffer, bodySection *imap.FetchItemBodySection) { + env := buf.Envelope + if env == nil { + slog.Warn("imap: message has no envelope, skipping") + return + } + + messageID := env.MessageID + subject := env.Subject + + // Extract sender email. + sender := "" + if len(env.From) > 0 { + sender = env.From[0].Addr() + } + + log := logging.TaskLogger("", "", messageID) + + // 1. Anti-loop: skip if sender is the system email. + if strings.EqualFold(sender, w.cfg.SystemEmail) { + log.Info("imap: anti-loop, ignoring own email", "sender", sender) + w.saveIgnored(messageID, sender, subject, "anti-loop: system email") + w.markSeen(c, buf.UID) + return + } + + // 2. Whitelist: skip if sender not in allowed list. + if !w.isWhitelisted(sender) { + log.Info("imap: sender not whitelisted, ignoring", "sender", sender) + w.saveIgnored(messageID, sender, subject, "not whitelisted") + w.markSeen(c, buf.UID) + return + } + + // 3. Idempotency: skip if message_id already exists. + existing, err := database.FindByMessageID(w.db, messageID) + if err != nil { + log.Error("imap: db lookup failed", "error", err) + return + } + if existing != nil { + log.Info("imap: duplicate message_id, skipping") + w.markSeen(c, buf.UID) + return + } + + // 4. Threading: determine thread_id. + threadID := messageID + inReplyTo := "" + + // Parse body to get In-Reply-To / References headers and plain text. + bodyPlain := "" + rawBody := buf.FindBodySection(bodySection) + if rawBody != nil { + mr, err := gomessage.CreateReader(io.NopCloser(strings.NewReader(string(rawBody)))) + if err == nil { + // Extract In-Reply-To header. + if irt, err := mr.Header.Text("In-Reply-To"); err == nil && irt != "" { + inReplyTo = strings.TrimSpace(irt) + } + + // Read body parts for plain text. + for { + p, err := mr.NextPart() + if err != nil { + break + } + if _, ok := p.Header.(*gomessage.InlineHeader); ok { + b, _ := io.ReadAll(p.Body) + if bodyPlain == "" { + bodyPlain = string(b) + } + } + } + } + } + + if inReplyTo != "" { + parent, err := database.FindByMessageID(w.db, inReplyTo) + if err == nil && parent != nil { + threadID = parent.ThreadID + } + } + + // 5. Save task with status RECEIVED. + taskUUID := uuid.New().String() + task := &database.Task{ + TaskUUID: taskUUID, + ThreadID: threadID, + MessageID: messageID, + Sender: sender, + Subject: subject, + BodyPlain: bodyPlain, + Status: database.StatusReceived, + } + + taskLog := logging.TaskLogger(taskUUID, threadID, messageID) + + if err := database.CreateTask(w.db, task); err != nil { + taskLog.Error("imap: failed to save task", "error", err) + return + } + + taskLog.Info("imap: task created", + "sender", sender, + "subject", subject, + "status", database.StatusReceived, + ) + + // Mark as seen in IMAP so we don't re-process. + w.markSeen(c, buf.UID) + + // Trigger dispatch (will be wired later). + if w.OnReceived != nil { + w.OnReceived(task) + } +} + +// saveIgnored records an ignored email in the DB for auditability. +func (w *IMAPWatcher) saveIgnored(messageID, sender, subject, reason string) { + taskUUID := uuid.New().String() + errMsg := reason + task := &database.Task{ + TaskUUID: taskUUID, + ThreadID: messageID, + MessageID: messageID, + Sender: sender, + Subject: subject, + Status: database.StatusIgnored, + LastError: &errMsg, + } + if err := database.CreateTask(w.db, task); err != nil { + slog.Error("imap: failed to save ignored task", "error", err, "message_id", messageID) + } +} + +// isWhitelisted checks if the sender is in the allowed list. +// Empty whitelist means all senders are allowed. +func (w *IMAPWatcher) isWhitelisted(sender string) bool { + if len(w.cfg.WhitelistEmails) == 0 { + return true + } + lower := strings.ToLower(sender) + for _, email := range w.cfg.WhitelistEmails { + if strings.ToLower(email) == lower { + return true + } + } + return false +} + +// markSeen flags a message as \Seen in IMAP. +func (w *IMAPWatcher) markSeen(c *imapclient.Client, uid imap.UID) { + storeFlags := imap.StoreFlags{ + Op: imap.StoreFlagsAdd, + Flags: []imap.Flag{imap.FlagSeen}, + Silent: true, + } + if err := c.Store(imap.UIDSetNum(uid), &storeFlags, nil).Close(); err != nil { + slog.Error("imap: failed to mark message as seen", "error", err, "uid", uid) + } +} diff --git a/internal/mail/smtp.go b/internal/mail/smtp.go new file mode 100644 index 0000000..138bd7b --- /dev/null +++ b/internal/mail/smtp.go @@ -0,0 +1,134 @@ +package mail + +import ( + "fmt" + "net/smtp" + "strings" + "time" + + "thuanle.me/claw-email-bridge/internal/config" + "thuanle.me/claw-email-bridge/internal/database" + "thuanle.me/claw-email-bridge/internal/logging" + "gorm.io/gorm" +) + +// Retry backoff schedule: 1s, 5s, 15s. +var smtpRetryBackoffs = []time.Duration{1 * time.Second, 5 * time.Second, 15 * time.Second} + +// SMTPSender sends reply emails via SMTP with retry logic. +type SMTPSender struct { + cfg *config.Config + db *gorm.DB +} + +// NewSMTPSender creates a new SMTP sender. +func NewSMTPSender(cfg *config.Config, db *gorm.DB) *SMTPSender { + return &SMTPSender{cfg: cfg, db: db} +} + +// SendReply sends an AI-generated reply to the original sender. +// This should be called asynchronously (go sender.SendReply(task)). +func (s *SMTPSender) SendReply(task *database.Task) { + log := logging.TaskLogger(task.TaskUUID, task.ThreadID, task.MessageID) + + // Build the email message with threading headers. + msg := s.buildMessage(task) + + for attempt := 1; attempt <= len(smtpRetryBackoffs); attempt++ { + task.AttemptSMTP = attempt + + log.Info("smtp: sending reply", + "attempt", attempt, + "to", task.Sender, + ) + + err := s.send(task.Sender, msg) + if err == nil { + task.Status = database.StatusCompleted + if err := database.UpdateTask(s.db, task); err != nil { + log.Error("smtp: failed to save COMPLETED status", "error", err) + } + log.Info("smtp: reply sent successfully", + "attempt", attempt, + "status", database.StatusCompleted, + ) + return + } + + log.Warn("smtp: send failed", + "attempt", attempt, + "error", err, + ) + + // Update status to SMTP_RETRYING. + task.Status = database.StatusSMTPRetrying + if err := database.UpdateTask(s.db, task); err != nil { + log.Error("smtp: failed to save retry status", "error", err) + } + + // Last attempt — mark as FAILED. + if attempt == len(smtpRetryBackoffs) { + errMsg := err.Error() + task.Status = database.StatusFailed + task.LastError = &errMsg + if err := database.UpdateTask(s.db, task); err != nil { + log.Error("smtp: failed to save FAILED status", "error", err) + } + log.Error("smtp: all retries exhausted", + "attempt", attempt, + "status", database.StatusFailed, + "error", errMsg, + ) + return + } + + backoff := smtpRetryBackoffs[attempt-1] + log.Info("smtp: retrying after backoff", "backoff", backoff) + time.Sleep(backoff) + } +} + +// buildMessage constructs the RFC 2822 email with threading headers. +func (s *SMTPSender) buildMessage(task *database.Task) string { + subject := task.Subject + if !strings.HasPrefix(strings.ToLower(subject), "re:") { + subject = "Re: " + subject + } + + var b strings.Builder + b.WriteString(fmt.Sprintf("From: %s\r\n", s.cfg.SystemEmail)) + b.WriteString(fmt.Sprintf("To: %s\r\n", task.Sender)) + b.WriteString(fmt.Sprintf("Subject: %s\r\n", subject)) + + // Threading headers — required by requirements 4.3. + b.WriteString(fmt.Sprintf("In-Reply-To: %s\r\n", task.MessageID)) + b.WriteString(fmt.Sprintf("References: %s\r\n", task.MessageID)) + + b.WriteString("MIME-Version: 1.0\r\n") + b.WriteString("Content-Type: text/plain; charset=UTF-8\r\n") + b.WriteString("\r\n") + b.WriteString(task.AIResponse) + + return b.String() +} + +// send performs the actual SMTP delivery. +func (s *SMTPSender) send(to string, msg string) error { + addr := fmt.Sprintf("%s:%s", s.cfg.SMTPHost, s.cfg.SMTPPort) + + auth := smtp.PlainAuth("", s.cfg.SMTPUser, s.cfg.SMTPPass, s.cfg.SMTPHost) + + err := smtp.SendMail(addr, auth, s.cfg.SystemEmail, []string{to}, []byte(msg)) + if err != nil { + return fmt.Errorf("smtp sendmail: %w", err) + } + + return nil +} + +// SendReplyFunc returns a function compatible with api.OnCallbackDone. +func (s *SMTPSender) SendReplyFunc() func(task *database.Task) { + return func(task *database.Task) { + s.SendReply(task) + } +} diff --git a/requirements.md b/requirements.md new file mode 100644 index 0000000..9ea3f7c --- /dev/null +++ b/requirements.md @@ -0,0 +1,178 @@ +PROJECT: EMAIL-OPENCLAW BRIDGE (STATEFUL, KISS) + +1) Mục tiêu & phạm vi + +Mục tiêu hệ thống: +- Tự động xử lý email bằng OpenClaw theo luồng: + Ingress (IMAP) -> Dispatch (OpenClaw) -> Callback -> Egress (SMTP reply). + +Phạm vi v1: +- Stateful bằng SQLite. +- Hỗ trợ email thread theo `thread_id`. +- Idempotency theo `message_id`. +- Webhook callback có xác thực bằng token đơn giản. +- Bổ sung health endpoints: + - `GET /healthz` (liveness) + - `GET /readyz` (readiness) +- Retry cơ bản (KISS) cho OpenClaw và SMTP: + - Tối đa 3 lần với backoff: 1s, 5s, 15s. + +Non-goals v1: +- Không rate limit. +- Không DLQ/reprocess UI. +- Không stack metrics nâng cao. + +2) Công nghệ yêu cầu + +- Ngôn ngữ: Go. +- Database: SQLite. + - Khuyến nghị: `gorm.io/gorm` + `modernc.org/sqlite` (không cần CGO khi build Docker). +- IMAP: `github.com/emersion/go-imap` (+ IDLE). +- SMTP: `net/smtp` hoặc thư viện gửi mail Go tương đương. +- HTTP/Webhook: `gin-gonic/gin` hoặc `gofiber/fiber`. +- Proxy IMAP (optional): `golang.org/x/net/proxy`. +- Triển khai: Docker + Docker Compose. + +3) Database schema + +Bảng `tasks`: +- `id` INTEGER PRIMARY KEY AUTOINCREMENT +- `task_uuid` TEXT UNIQUE +- `thread_id` TEXT +- `message_id` TEXT +- `sender` TEXT +- `subject` TEXT +- `body_plain` TEXT +- `ai_response` TEXT +- `status` TEXT +- `attempt_openclaw` INTEGER DEFAULT 0 +- `attempt_smtp` INTEGER DEFAULT 0 +- `last_error` TEXT NULL +- `next_retry_at` DATETIME NULL (chỉ cần nếu retry xử lý theo scheduler/worker) +- `created_at` DATETIME +- `updated_at` DATETIME + +Danh sách trạng thái hợp lệ: +- `IGNORED` +- `RECEIVED` +- `AI_PROCESSING` +- `CALLBACK_DONE` +- `SMTP_RETRYING` +- `COMPLETED` +- `FAILED` + +4) Luồng xử lý chi tiết + +4.1 Ingress (IMAP) +- Theo dõi mailbox qua IMAP IDLE. +- Nếu có `IMAP_PROXY_URL` thì kết nối IMAP qua proxy; nếu không thì direct. +- Khi có email mới: + 1. Anti-loop: nếu `sender` trùng email hệ thống -> `IGNORED`. + 2. Whitelist: nếu không nằm trong danh sách cho phép -> `IGNORED`. + 3. Idempotency: nếu `message_id` đã tồn tại -> bỏ qua. + 4. Threading: + - Nếu có `In-Reply-To`/`References`: tìm email cha trong DB, kế thừa `thread_id`. + - Nếu không có: `thread_id = message_id`. + 5. Lưu DB với `status = RECEIVED`. + +4.2 Dispatch (OpenClaw) +- Lấy nội dung email hiện tại và history của cùng `thread_id` (chỉ các bản ghi `COMPLETED`). +- Gọi OpenClaw API bằng HTTP client đi direct (không kế thừa proxy hệ thống). +- Payload tối thiểu: + - `input` + - `session_id` = `thread_id` + - `history` (optional) + - `callback_url` + - `metadata.task_uuid` +- Cập nhật `status = AI_PROCESSING`. + +Retry OpenClaw: +- Khi lỗi tạm thời: retry tối đa 3 lần theo backoff 1s, 5s, 15s. +- Mỗi lần retry tăng `attempt_openclaw`. +- Hết retry: `status = FAILED`, ghi `last_error`. + +4.3 Callback + Egress (SMTP) +- Webhook nhận callback từ OpenClaw. +- Tìm task theo `metadata.task_uuid`. +- Nếu task đã `COMPLETED`: trả 200 và không cập nhật gì thêm (idempotent callback). +- Nếu hợp lệ: cập nhật `ai_response`, `status = CALLBACK_DONE`. + +Gửi email phản hồi: +- Subject/Body theo kết quả AI. +- Bắt buộc set `In-Reply-To` và `References` khớp message liên quan để giữ thread phía người nhận. + +Retry SMTP: +- Nếu lỗi tạm thời: chuyển `SMTP_RETRYING`, retry 3 lần theo 1s, 5s, 15s. +- Mỗi lần retry tăng `attempt_smtp`. +- Hết retry: `status = FAILED`, ghi `last_error`. +- Thành công: `status = COMPLETED`. + +5) API contract & bảo mật + +5.1 Callback contract (từ OpenClaw) +Required: +- `metadata.task_uuid` +- `result` (text) +- `status` + +5.2 Xác thực webhook (KISS) +- Yêu cầu header: `X-Bridge-Token`. +- Giá trị kỳ vọng lấy từ `.env`, ví dụ: `BRIDGE_CALLBACK_TOKEN`. +- Thiếu/sai token -> trả `401 Unauthorized`. + +5.3 Health endpoints +- `GET /healthz` -> `200 {"status":"ok"}` nếu process sống. +- `GET /readyz`: + - kiểm tra DB writable, + - kiểm tra cấu hình bắt buộc có đủ (IMAP/SMTP/OpenClaw/token), + - nếu lỗi bất kỳ check nào -> `503`. + +6) Logging & observability (mức tối thiểu) + +- Structured logging ra stdout. +- Mỗi log liên quan xử lý email phải có: + - `task_uuid` + - `thread_id` + - `message_id` + - `status` + - `attempt` + - `error` (nếu có) + +7) Cấu hình môi trường + +Đọc từ `.env`: +- `IMAP_HOST`, `IMAP_PORT`, `IMAP_USER`, `IMAP_PASS` +- `SMTP_HOST`, `SMTP_PORT`, `SMTP_USER`, `SMTP_PASS` +- `OPENCLAW_URL`, `OPENCLAW_API_KEY` (nếu dùng) +- `BRIDGE_CALLBACK_TOKEN` +- `WHITELIST_EMAILS` (comma-separated) +- `SYSTEM_EMAIL` +- `IMAP_PROXY_URL` (optional) + +Ví dụ: +- `WHITELIST_EMAILS="user1@example.com,user2@example.com"` +- `IMAP_PROXY_URL="socks5://user:pass@proxy-host:port"` + +8) Docker triển khai + +- Multi-stage build để tối ưu image size. +- `docker-compose.yml` mount volume cho SQLite DB (vd `database.db`). +- Xem log bằng: `docker logs -f `. + +9) Acceptance criteria (bắt buộc đạt) + +1. Email hợp lệ đi hết pipeline và kết thúc `COMPLETED`. +2. Thread reply giữ đúng `In-Reply-To`/`References`. +3. OpenClaw/SMTP retry đúng thứ tự 1s -> 5s -> 15s. +4. Hết retry thì `FAILED` và có `last_error`. +5. Callback sai token trả 401 và không đổi state task. +6. Anti-loop + whitelist + idempotency hoạt động đúng. + +10) Test matrix tối thiểu + +1. Happy path: mail mới -> callback hợp lệ -> SMTP thành công -> `COMPLETED`. +2. OpenClaw fail 2 lần, lần 3 thành công -> `COMPLETED`, `attempt_openclaw = 3`. +3. SMTP fail hết retry -> `FAILED`, có `last_error`. +4. Callback sai token -> 401, state không đổi. +5. Duplicate `message_id` -> không tạo task mới. +6. Sender là chính hệ thống hoặc ngoài whitelist -> `IGNORED`.