From 319a2fa689f95ad6e5962d1dfa6138168fad399f Mon Sep 17 00:00:00 2001 From: karti-ai Date: Sun, 21 Jun 2026 12:54:25 -0700 Subject: [PATCH] Milestone 1: core services + Mox ingest + wired v0 API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - internal/mail: MailBackend interface (Send/Start/Capabilities) + NullBackend. - internal/core: inbox/message/thread services over pgx; default-pod bootstrap. - IngestRaw: real mox/message parse → body extraction (text/html walk) → quote-stripping (extracted_text) → reference-based threading → store, all in one tx; emits message.received event; populates FTS tsvector. - core implements mail.InboundSink (Deliver) so any backend feeds the same path. - internal/api: v0 routes wired live (was 501) — create/list/get inbox, ingest, list/get messages, list/get thread(+messages), send/reply (503 via NullBackend). - cmd/openmail serve: builds core + NullBackend, ensures default pod. Verified e2e vs Postgres 16: inbox create; ingest original+reply → same thread via In-Reply-To/References; extracted_text strips quoted history; thread detail ordered; FTS matches 'invoice'; events written; send → honest 503. vet clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- cmd/openmail/main.go | 15 +- internal/api/handlers.go | 244 ++++++++++++++++++++++++++++ internal/api/server.go | 70 +++++--- internal/core/core.go | 58 +++++++ internal/core/inbox.go | 72 ++++++++ internal/core/message.go | 343 +++++++++++++++++++++++++++++++++++++++ internal/core/thread.go | 54 ++++++ internal/mail/backend.go | 77 +++++++++ 8 files changed, 906 insertions(+), 27 deletions(-) create mode 100644 internal/api/handlers.go create mode 100644 internal/core/core.go create mode 100644 internal/core/inbox.go create mode 100644 internal/core/message.go create mode 100644 internal/core/thread.go create mode 100644 internal/mail/backend.go diff --git a/cmd/openmail/main.go b/cmd/openmail/main.go index 66f4b30..18220a4 100644 --- a/cmd/openmail/main.go +++ b/cmd/openmail/main.go @@ -16,6 +16,8 @@ import ( "github.com/karti-ai/openmail/internal/api" "github.com/karti-ai/openmail/internal/config" + "github.com/karti-ai/openmail/internal/core" + "github.com/karti-ai/openmail/internal/mail" "github.com/karti-ai/openmail/internal/store" ) @@ -110,15 +112,26 @@ func runServe(ctx context.Context, cfg config.Config) error { if err != nil { return err } + + var svc *core.Service + var podID string if st != nil { defer st.Close() + svc = core.New(st) + if podID, err = svc.EnsureDefaultPod(ctx); err != nil { + return fmt.Errorf("serve: ensure default pod: %w", err) + } } else { fmt.Fprintln(os.Stderr, "warning: DATABASE_URL unset — serving health only, API will report db not configured") } + // Milestone 1: NullBackend (mail enters only via the ingest API; sending is + // unavailable until a relay/imap_smtp/embedded backend is wired). + backend := mail.NullBackend{} + srv := &http.Server{ Addr: cfg.HTTPAddr, - Handler: api.New(cfg, st).Router(), + Handler: api.New(cfg, svc, backend, podID).Router(), ReadHeaderTimeout: 10 * time.Second, } diff --git a/internal/api/handlers.go b/internal/api/handlers.go new file mode 100644 index 0000000..e7012f5 --- /dev/null +++ b/internal/api/handlers.go @@ -0,0 +1,244 @@ +package api + +import ( + "encoding/json" + "errors" + "io" + "net/http" + "strconv" + + "github.com/go-chi/chi/v5" + + "github.com/karti-ai/openmail/internal/core" + "github.com/karti-ai/openmail/internal/mail" +) + +const maxIngestBytes = 30 << 20 // 30 MiB raw message cap + +func (s *Server) createInbox(w http.ResponseWriter, r *http.Request) { + if !s.requireCore(w) { + return + } + var body struct { + Address string `json:"address"` + DisplayName *string `json:"display_name"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Address == "" { + writeErr(w, http.StatusBadRequest, "invalid_request", "address is required") + return + } + ib, err := s.core.CreateInbox(r.Context(), s.podID, body.Address, body.DisplayName) + if err != nil { + writeErr(w, http.StatusInternalServerError, "create_failed", err.Error()) + return + } + writeJSON(w, http.StatusCreated, ib) +} + +func (s *Server) listInboxes(w http.ResponseWriter, r *http.Request) { + if !s.requireCore(w) { + return + } + list, err := s.core.ListInboxes(r.Context(), s.podID) + if err != nil { + writeErr(w, http.StatusInternalServerError, "list_failed", err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"inboxes": list}) +} + +func (s *Server) getInbox(w http.ResponseWriter, r *http.Request) { + if !s.requireCore(w) { + return + } + ib, err := s.core.GetInbox(r.Context(), chi.URLParam(r, "id")) + if handleErr(w, err) { + return + } + writeJSON(w, http.StatusOK, ib) +} + +// ingest accepts a raw RFC 5322 message body and stores it in the inbox — the +// milestone-1 seed path, and the same path a MailBackend uses for real inbound. +func (s *Server) ingest(w http.ResponseWriter, r *http.Request) { + if !s.requireCore(w) { + return + } + inboxID := chi.URLParam(r, "id") + if _, err := s.core.GetInbox(r.Context(), inboxID); handleErr(w, err) { + return + } + raw, err := io.ReadAll(io.LimitReader(r.Body, maxIngestBytes)) + if err != nil || len(raw) == 0 { + writeErr(w, http.StatusBadRequest, "invalid_request", "raw message body required") + return + } + msg, err := s.core.IngestRaw(r.Context(), inboxID, raw) + if err != nil { + writeErr(w, http.StatusInternalServerError, "ingest_failed", err.Error()) + return + } + writeJSON(w, http.StatusCreated, msg) +} + +func (s *Server) listMessages(w http.ResponseWriter, r *http.Request) { + if !s.requireCore(w) { + return + } + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) + list, err := s.core.ListMessages(r.Context(), chi.URLParam(r, "id"), limit) + if err != nil { + writeErr(w, http.StatusInternalServerError, "list_failed", err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"messages": list}) +} + +func (s *Server) getMessage(w http.ResponseWriter, r *http.Request) { + if !s.requireCore(w) { + return + } + m, err := s.core.GetMessage(r.Context(), chi.URLParam(r, "msgID")) + if handleErr(w, err) { + return + } + writeJSON(w, http.StatusOK, m) +} + +func (s *Server) listThreads(w http.ResponseWriter, r *http.Request) { + if !s.requireCore(w) { + return + } + list, err := s.core.ListThreads(r.Context(), chi.URLParam(r, "id")) + if err != nil { + writeErr(w, http.StatusInternalServerError, "list_failed", err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"threads": list}) +} + +func (s *Server) getThread(w http.ResponseWriter, r *http.Request) { + if !s.requireCore(w) { + return + } + threadID := chi.URLParam(r, "threadID") + t, err := s.core.GetThread(r.Context(), threadID) + if handleErr(w, err) { + return + } + msgs, err := s.core.GetThreadMessages(r.Context(), threadID) + if err != nil { + writeErr(w, http.StatusInternalServerError, "list_failed", err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"thread": t, "messages": msgs}) +} + +func (s *Server) sendMessage(w http.ResponseWriter, r *http.Request) { + if !s.requireCore(w) { + return + } + inboxID := chi.URLParam(r, "id") + ib, err := s.core.GetInbox(r.Context(), inboxID) + if handleErr(w, err) { + return + } + var body struct { + To []string `json:"to"` + Cc []string `json:"cc"` + Bcc []string `json:"bcc"` + Subject string `json:"subject"` + Text string `json:"text"` + HTML string `json:"html"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil || len(body.To) == 0 { + writeErr(w, http.StatusBadRequest, "invalid_request", "at least one 'to' recipient is required") + return + } + s.dispatch(w, r, &mail.OutgoingMessage{ + InboxID: inboxID, From: ib.Address, + To: body.To, Cc: body.Cc, Bcc: body.Bcc, + Subject: body.Subject, Text: body.Text, HTML: body.HTML, + }) +} + +func (s *Server) replyMessage(w http.ResponseWriter, r *http.Request) { + if !s.requireCore(w) { + return + } + inboxID := chi.URLParam(r, "id") + ib, err := s.core.GetInbox(r.Context(), inboxID) + if handleErr(w, err) { + return + } + orig, err := s.core.GetMessage(r.Context(), chi.URLParam(r, "msgID")) + if handleErr(w, err) { + return + } + var body struct { + Text string `json:"text"` + HTML string `json:"html"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeErr(w, http.StatusBadRequest, "invalid_request", "invalid body") + return + } + out := &mail.OutgoingMessage{ + InboxID: inboxID, From: ib.Address, + Subject: replySubject(orig.Subject), + Text: body.Text, HTML: body.HTML, + References: append(append([]string{}, orig.References...), strPtr(orig.MessageIDHdr)...), + } + if orig.MessageIDHdr != nil { + out.InReplyTo = *orig.MessageIDHdr + } + if orig.FromAddr != nil { + out.To = []string{*orig.FromAddr} + } + s.dispatch(w, r, out) +} + +// dispatch hands an outgoing message to the active backend, translating the +// NullBackend's ErrNotSupported into a clear 503 (no send path configured yet). +func (s *Server) dispatch(w http.ResponseWriter, r *http.Request, out *mail.OutgoingMessage) { + res, err := s.backend.Send(r.Context(), out) + if errors.Is(err, mail.ErrNotSupported) { + writeErr(w, http.StatusServiceUnavailable, "send_unavailable", + "no send-capable mail backend configured (relay/imap_smtp/embedded land in later milestones)") + return + } + if err != nil { + writeErr(w, http.StatusBadGateway, "send_failed", err.Error()) + return + } + writeJSON(w, http.StatusAccepted, res) +} + +func handleErr(w http.ResponseWriter, err error) bool { + switch { + case err == nil: + return false + case errors.Is(err, core.ErrNotFound): + writeErr(w, http.StatusNotFound, "not_found", "resource not found") + default: + writeErr(w, http.StatusInternalServerError, "internal_error", err.Error()) + } + return true +} + +func replySubject(s *string) string { + if s == nil || *s == "" { + return "Re:" + } + if len(*s) >= 3 && (*s)[:3] == "Re:" { + return *s + } + return "Re: " + *s +} + +func strPtr(s *string) []string { + if s == nil || *s == "" { + return nil + } + return []string{*s} +} diff --git a/internal/api/server.go b/internal/api/server.go index 93adfc0..6904310 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -1,6 +1,5 @@ // Package api is OpenMail's agent-facing HTTP surface: the AgentMail-shaped v0 -// REST API (see ARCHITECTURE.md §4). v0 routes are stubbed pending the core -// services; health and auth are real. +// REST API (see ARCHITECTURE.md §4). package api import ( @@ -11,16 +10,19 @@ import ( "github.com/go-chi/chi/v5/middleware" "github.com/karti-ai/openmail/internal/config" - "github.com/karti-ai/openmail/internal/store" + "github.com/karti-ai/openmail/internal/core" + "github.com/karti-ai/openmail/internal/mail" ) type Server struct { - cfg config.Config - store *store.Store + cfg config.Config + core *core.Service // nil when DATABASE_URL is unset + backend mail.MailBackend // never nil (NullBackend by default) + podID string // default pod, resolved at startup } -func New(cfg config.Config, st *store.Store) *Server { - return &Server{cfg: cfg, store: st} +func New(cfg config.Config, svc *core.Service, backend mail.MailBackend, podID string) *Server { + return &Server{cfg: cfg, core: svc, backend: backend, podID: podID} } func (s *Server) Router() http.Handler { @@ -29,23 +31,22 @@ func (s *Server) Router() http.Handler { r.Use(middleware.RealIP) r.Use(middleware.Recoverer) - // Unauthenticated liveness/readiness. r.Get("/healthz", s.handleHealth) - // Authenticated v0 surface. r.Group(func(r chi.Router) { r.Use(s.bearerAuth) r.Route("/v0/inboxes", func(r chi.Router) { - r.Post("/", notImplemented) // create inbox - r.Get("/", notImplemented) // list inboxes - r.Get("/{id}", notImplemented) // get inbox - r.Post("/{id}/messages/send", notImplemented) - r.Get("/{id}/messages", notImplemented) - r.Get("/{id}/messages/{msgID}", notImplemented) - r.Post("/{id}/messages/{msgID}/reply", notImplemented) - r.Get("/{id}/threads", notImplemented) - r.Get("/{id}/threads/{threadID}", notImplemented) + r.Post("/", s.createInbox) + r.Get("/", s.listInboxes) + r.Get("/{id}", s.getInbox) + r.Post("/{id}/ingest", s.ingest) // seed/inbound path (milestone 1) + r.Post("/{id}/messages/send", s.sendMessage) + r.Get("/{id}/messages", s.listMessages) + r.Get("/{id}/messages/{msgID}", s.getMessage) + r.Post("/{id}/messages/{msgID}/reply", s.replyMessage) + r.Get("/{id}/threads", s.listThreads) + r.Get("/{id}/threads/{threadID}", s.getThread) }) }) @@ -53,9 +54,9 @@ func (s *Server) Router() http.Handler { } func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { - status := map[string]any{"status": "ok"} - if s.store != nil { - if err := s.store.Pool.Ping(r.Context()); err != nil { + status := map[string]any{"status": "ok", "backend": backendName(s.backend)} + if s.core != nil { + if err := s.core.Ping(r.Context()); err != nil { status["status"] = "degraded" status["db"] = err.Error() writeJSON(w, http.StatusServiceUnavailable, status) @@ -68,11 +69,24 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, status) } -func notImplemented(w http.ResponseWriter, r *http.Request) { - writeJSON(w, http.StatusNotImplemented, map[string]string{ - "error": "not_implemented", - "message": "endpoint scaffolded; core service pending (milestone 1)", - }) +func backendName(b mail.MailBackend) string { + if _, ok := b.(mail.NullBackend); ok { + return "null" + } + return "configured" +} + +// requireCore guards handlers that need the database. Returns false (and writes +// a 503) when the store is unconfigured. +func (s *Server) requireCore(w http.ResponseWriter) bool { + if s.core == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{ + "error": "db_unconfigured", + "message": "DATABASE_URL not set", + }) + return false + } + return true } func writeJSON(w http.ResponseWriter, code int, v any) { @@ -80,3 +94,7 @@ func writeJSON(w http.ResponseWriter, code int, v any) { w.WriteHeader(code) _ = json.NewEncoder(w).Encode(v) } + +func writeErr(w http.ResponseWriter, code int, errCode, msg string) { + writeJSON(w, code, map[string]string{"error": errCode, "message": msg}) +} diff --git a/internal/core/core.go b/internal/core/core.go new file mode 100644 index 0000000..ae015ae --- /dev/null +++ b/internal/core/core.go @@ -0,0 +1,58 @@ +// Package core holds OpenMail's domain services: inboxes, messages, threads, +// drafts, and the inbound ingest path (parse → thread → store). It is the layer +// the API and MCP server call, and it implements mail.InboundSink so any +// MailBackend can deliver into it. +package core + +import ( + "context" + "errors" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/karti-ai/openmail/internal/store" +) + +// ErrNotFound is returned when a requested resource does not exist. +var ErrNotFound = errors.New("core: not found") + +type Service struct { + pool *pgxpool.Pool +} + +func New(st *store.Store) *Service { return &Service{pool: st.Pool} } + +// Ping verifies the database connection (used by the health endpoint). +func (s *Service) Ping(ctx context.Context) error { return s.pool.Ping(ctx) } + +// EnsureDefaultPod returns the id of the singleton "default" pod, creating it if +// absent. Until DB-backed api_keys carry a pod_id, the API operates within this +// one tenant. Idempotent. +func (s *Service) EnsureDefaultPod(ctx context.Context) (string, error) { + var id string + err := s.pool.QueryRow(ctx, `SELECT id::text FROM pods WHERE name = 'default'`).Scan(&id) + if err == nil { + return id, nil + } + if !errors.Is(err, pgx.ErrNoRows) { + return "", err + } + err = s.pool.QueryRow(ctx, + `INSERT INTO pods (name) VALUES ('default') + ON CONFLICT DO NOTHING + RETURNING id::text`).Scan(&id) + if errors.Is(err, pgx.ErrNoRows) { + // Lost a race; read the row the other writer created. + err = s.pool.QueryRow(ctx, `SELECT id::text FROM pods WHERE name = 'default'`).Scan(&id) + } + return id, err +} + +// nullStr maps an optional string to a value usable as a nullable SQL arg. +func nullStr(s *string) any { + if s == nil { + return nil + } + return *s +} diff --git a/internal/core/inbox.go b/internal/core/inbox.go new file mode 100644 index 0000000..0ff56e7 --- /dev/null +++ b/internal/core/inbox.go @@ -0,0 +1,72 @@ +package core + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/jackc/pgx/v5" +) + +type Inbox struct { + ID string `json:"id"` + PodID string `json:"pod_id"` + Address string `json:"address"` + DisplayName *string `json:"display_name,omitempty"` + Metadata json.RawMessage `json:"metadata,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +const inboxCols = `id::text, pod_id::text, address, display_name, metadata, created_at, updated_at` + +func scanInbox(row pgx.Row) (Inbox, error) { + var ib Inbox + err := row.Scan(&ib.ID, &ib.PodID, &ib.Address, &ib.DisplayName, &ib.Metadata, &ib.CreatedAt, &ib.UpdatedAt) + return ib, err +} + +// CreateInbox provisions a new agent-owned address within a pod. +func (s *Service) CreateInbox(ctx context.Context, podID, address string, displayName *string) (Inbox, error) { + row := s.pool.QueryRow(ctx, + `INSERT INTO inboxes (pod_id, address, display_name) + VALUES ($1, $2, $3) + RETURNING `+inboxCols, + podID, address, nullStr(displayName)) + return scanInbox(row) +} + +func (s *Service) GetInbox(ctx context.Context, id string) (Inbox, error) { + ib, err := scanInbox(s.pool.QueryRow(ctx, `SELECT `+inboxCols+` FROM inboxes WHERE id = $1`, id)) + if errors.Is(err, pgx.ErrNoRows) { + return Inbox{}, ErrNotFound + } + return ib, err +} + +func (s *Service) GetInboxByAddress(ctx context.Context, address string) (Inbox, error) { + ib, err := scanInbox(s.pool.QueryRow(ctx, `SELECT `+inboxCols+` FROM inboxes WHERE address = $1`, address)) + if errors.Is(err, pgx.ErrNoRows) { + return Inbox{}, ErrNotFound + } + return ib, err +} + +func (s *Service) ListInboxes(ctx context.Context, podID string) ([]Inbox, error) { + rows, err := s.pool.Query(ctx, + `SELECT `+inboxCols+` FROM inboxes WHERE pod_id = $1 ORDER BY created_at DESC`, podID) + if err != nil { + return nil, err + } + defer rows.Close() + out := []Inbox{} + for rows.Next() { + ib, err := scanInbox(rows) + if err != nil { + return nil, err + } + out = append(out, ib) + } + return out, rows.Err() +} diff --git a/internal/core/message.go b/internal/core/message.go new file mode 100644 index 0000000..2833e8d --- /dev/null +++ b/internal/core/message.go @@ -0,0 +1,343 @@ +package core + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "log/slog" + "regexp" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/mjl-/mox/message" +) + +type Message struct { + ID string `json:"id"` + InboxID string `json:"inbox_id"` + ThreadID *string `json:"thread_id,omitempty"` + MessageIDHdr *string `json:"message_id,omitempty"` + InReplyTo *string `json:"in_reply_to,omitempty"` + References []string `json:"references"` + FromAddr *string `json:"from,omitempty"` + ToAddrs []string `json:"to"` + Cc []string `json:"cc"` + Bcc []string `json:"bcc"` + Subject *string `json:"subject,omitempty"` + Preview *string `json:"preview,omitempty"` + Text *string `json:"text,omitempty"` + HTML *string `json:"html,omitempty"` + ExtractedText *string `json:"extracted_text,omitempty"` + Labels []string `json:"labels"` + SizeBytes *int64 `json:"size_bytes,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + +const messageCols = `id::text, inbox_id::text, thread_id::text, message_id_hdr, in_reply_to, ` + + `"references", from_addr, to_addrs, cc, bcc, subject, preview, text, html, extracted_text, ` + + `labels, size_bytes, created_at` + +func scanMessage(row pgx.Row) (Message, error) { + var m Message + err := row.Scan(&m.ID, &m.InboxID, &m.ThreadID, &m.MessageIDHdr, &m.InReplyTo, + &m.References, &m.FromAddr, &m.ToAddrs, &m.Cc, &m.Bcc, &m.Subject, &m.Preview, + &m.Text, &m.HTML, &m.ExtractedText, &m.Labels, &m.SizeBytes, &m.CreatedAt) + return m, err +} + +func (s *Service) GetMessage(ctx context.Context, id string) (Message, error) { + m, err := scanMessage(s.pool.QueryRow(ctx, `SELECT `+messageCols+` FROM messages WHERE id = $1`, id)) + if errors.Is(err, pgx.ErrNoRows) { + return Message{}, ErrNotFound + } + return m, err +} + +func (s *Service) ListMessages(ctx context.Context, inboxID string, limit int) ([]Message, error) { + if limit <= 0 || limit > 200 { + limit = 50 + } + rows, err := s.pool.Query(ctx, + `SELECT `+messageCols+` FROM messages WHERE inbox_id = $1 ORDER BY created_at DESC LIMIT $2`, + inboxID, limit) + if err != nil { + return nil, err + } + defer rows.Close() + return collectMessages(rows) +} + +func (s *Service) GetThreadMessages(ctx context.Context, threadID string) ([]Message, error) { + rows, err := s.pool.Query(ctx, + `SELECT `+messageCols+` FROM messages WHERE thread_id = $1 ORDER BY created_at ASC`, threadID) + if err != nil { + return nil, err + } + defer rows.Close() + return collectMessages(rows) +} + +func collectMessages(rows pgx.Rows) ([]Message, error) { + out := []Message{} + for rows.Next() { + m, err := scanMessage(rows) + if err != nil { + return nil, err + } + out = append(out, m) + } + return out, rows.Err() +} + +// Deliver implements mail.InboundSink: resolve the inbox by address and ingest. +func (s *Service) Deliver(ctx context.Context, inboxAddr string, raw []byte) error { + ib, err := s.GetInboxByAddress(ctx, inboxAddr) + if err != nil { + return err + } + _, err = s.IngestRaw(ctx, ib.ID, raw) + return err +} + +// IngestRaw parses a raw RFC 5322 message with mox, resolves its thread, and +// stores it — the single inbound path shared by the ingest API and every +// MailBackend. Runs in one transaction. +func (s *Service) IngestRaw(ctx context.Context, inboxID string, raw []byte) (Message, error) { + pr, err := parseRaw(raw) + if err != nil { + return Message{}, err + } + extracted := stripQuotes(pr.text) + preview := makePreview(extracted, pr.text) + headersJSON, _ := json.Marshal(pr.headers) + + tx, err := s.pool.Begin(ctx) + if err != nil { + return Message{}, err + } + defer tx.Rollback(ctx) + + threadID, err := resolveThreadTx(ctx, tx, inboxID, pr) + if err != nil { + return Message{}, err + } + + var msgID string + err = tx.QueryRow(ctx, + `INSERT INTO messages + (inbox_id, thread_id, message_id_hdr, in_reply_to, "references", from_addr, + to_addrs, cc, bcc, subject, preview, text, html, extracted_text, + size_bytes, headers, ts) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16, + to_tsvector('english', coalesce($10,'') || ' ' || coalesce($12,''))) + RETURNING id::text`, + inboxID, threadID, nullEmpty(pr.messageID), nullEmpty(pr.inReplyTo), pr.references, + nullEmpty(pr.from), pr.to, pr.cc, pr.bcc, nullEmpty(pr.subject), nullEmpty(preview), + nullEmpty(pr.text), nullEmpty(pr.html), nullEmpty(extracted), + int64(len(raw)), headersJSON, + ).Scan(&msgID) + if err != nil { + return Message{}, err + } + + if _, err := tx.Exec(ctx, + `UPDATE threads + SET message_count = message_count + 1, + last_message_id = $1, + subject = COALESCE(subject, $2), + updated_at = now() + WHERE id = $3`, + msgID, nullEmpty(pr.subject), threadID); err != nil { + return Message{}, err + } + + evt, _ := json.Marshal(map[string]string{"message_id": msgID, "thread_id": threadID}) + if _, err := tx.Exec(ctx, + `INSERT INTO events (inbox_id, type, payload) VALUES ($1, 'message.received', $2)`, + inboxID, evt); err != nil { + return Message{}, err + } + + if err := tx.Commit(ctx); err != nil { + return Message{}, err + } + return s.GetMessage(ctx, msgID) +} + +// resolveThreadTx finds the thread for a message via In-Reply-To/References +// (the correct, false-merge-safe mechanism), creating a new thread otherwise. +// Subject-based fallback is intentionally deferred (see ARCHITECTURE.md §9). +func resolveThreadTx(ctx context.Context, tx pgx.Tx, inboxID string, pr parsed) (string, error) { + cand := make([]string, 0, len(pr.references)+1) + if pr.inReplyTo != "" { + cand = append(cand, pr.inReplyTo) + } + cand = append(cand, pr.references...) + + if len(cand) > 0 { + var tid string + err := tx.QueryRow(ctx, + `SELECT thread_id::text FROM messages + WHERE inbox_id = $1 AND message_id_hdr = ANY($2) AND thread_id IS NOT NULL + ORDER BY created_at DESC LIMIT 1`, + inboxID, cand).Scan(&tid) + if err == nil { + return tid, nil + } + if !errors.Is(err, pgx.ErrNoRows) { + return "", err + } + } + + var tid string + err := tx.QueryRow(ctx, + `INSERT INTO threads (inbox_id, subject) VALUES ($1, $2) RETURNING id::text`, + inboxID, nullEmpty(pr.subject)).Scan(&tid) + return tid, err +} + +// --- parsing (mox/message) --- + +// parsed is the normalized result of parsing a raw message. +type parsed struct { + messageID string + inReplyTo string + references []string + from string + to []string + cc []string + bcc []string + subject string + text string + html string + headers map[string][]string +} + +var discardLog = slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError})) + +func parseRaw(raw []byte) (parsed, error) { + var pr parsed + p, err := message.EnsurePart(discardLog, false, bytes.NewReader(raw), int64(len(raw))) + if err != nil { + return pr, err + } + if p.Envelope != nil { + e := p.Envelope + pr.subject = e.Subject + pr.messageID = e.MessageID + pr.inReplyTo = e.InReplyTo + pr.from = firstAddr(e.From) + pr.to = addrList(e.To) + pr.cc = addrList(e.CC) + pr.bcc = addrList(e.BCC) + } + if hdr, herr := p.Header(); herr == nil { + pr.references = strings.Fields(hdr.Get("References")) + pr.headers = hdr + } + if pr.references == nil { + pr.references = []string{} + } + pr.text, pr.html = extractBodies(&p) + return pr, nil +} + +// extractBodies walks the MIME tree and returns the first text/plain and +// text/html leaf bodies (decoded UTF-8). +func extractBodies(p *message.Part) (text, html string) { + if len(p.Parts) == 0 { + body := readBody(p) + switch { + case p.MediaType == "TEXT" && p.MediaSubType == "HTML": + return "", body + case p.MediaType == "TEXT" || p.MediaType == "": + return body, "" // PLAIN, or absent content-type → treat as plain + default: + return "", "" + } + } + for i := range p.Parts { + t, h := extractBodies(&p.Parts[i]) + if text == "" { + text = t + } + if html == "" { + html = h + } + } + return text, html +} + +const maxBodyBytes = 2 << 20 // 2 MiB cap per body part for milestone 1 + +func readBody(p *message.Part) string { + rd := p.ReaderUTF8OrBinary() + if rd == nil { + return "" + } + var b strings.Builder + _, _ = io.Copy(&b, io.LimitReader(rd, maxBodyBytes)) + return b.String() +} + +func firstAddr(as []message.Address) string { + if len(as) == 0 { + return "" + } + return as[0].User + "@" + as[0].Host +} + +func addrList(as []message.Address) []string { + out := make([]string, 0, len(as)) + for _, a := range as { + out = append(out, a.User+"@"+a.Host) + } + return out +} + +// --- text helpers --- + +var onWroteRe = regexp.MustCompile(`(?i)^on .+wrote:$`) + +// stripQuotes removes quoted history so an agent reads only the new content. +// Milestone-1 heuristic (talon-style port deferred): cut at the first quoted +// block or "On … wrote:" attribution line. +func stripQuotes(text string) string { + if text == "" { + return "" + } + lines := strings.Split(text, "\n") + out := make([]string, 0, len(lines)) + for _, ln := range lines { + t := strings.TrimSpace(ln) + if strings.HasPrefix(t, ">") || onWroteRe.MatchString(t) { + break + } + out = append(out, ln) + } + return strings.TrimSpace(strings.Join(out, "\n")) +} + +func makePreview(extracted, full string) string { + src := extracted + if src == "" { + src = full + } + src = strings.Join(strings.Fields(src), " ") + const max = 200 + if len([]rune(src)) > max { + src = string([]rune(src)[:max]) + } + return src +} + +// nullEmpty maps "" to a SQL NULL so optional text columns stay null, not blank. +func nullEmpty(s string) any { + if s == "" { + return nil + } + return s +} diff --git a/internal/core/thread.go b/internal/core/thread.go new file mode 100644 index 0000000..82b703a --- /dev/null +++ b/internal/core/thread.go @@ -0,0 +1,54 @@ +package core + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" +) + +type Thread struct { + ID string `json:"id"` + InboxID string `json:"inbox_id"` + Subject *string `json:"subject,omitempty"` + LastMessageID *string `json:"last_message_id,omitempty"` + MessageCount int `json:"message_count"` + Labels []string `json:"labels"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +const threadCols = `id::text, inbox_id::text, subject, last_message_id::text, message_count, labels, created_at, updated_at` + +func scanThread(row pgx.Row) (Thread, error) { + var t Thread + err := row.Scan(&t.ID, &t.InboxID, &t.Subject, &t.LastMessageID, &t.MessageCount, &t.Labels, &t.CreatedAt, &t.UpdatedAt) + return t, err +} + +func (s *Service) ListThreads(ctx context.Context, inboxID string) ([]Thread, error) { + rows, err := s.pool.Query(ctx, + `SELECT `+threadCols+` FROM threads WHERE inbox_id = $1 ORDER BY updated_at DESC`, inboxID) + if err != nil { + return nil, err + } + defer rows.Close() + out := []Thread{} + for rows.Next() { + t, err := scanThread(rows) + if err != nil { + return nil, err + } + out = append(out, t) + } + return out, rows.Err() +} + +func (s *Service) GetThread(ctx context.Context, id string) (Thread, error) { + t, err := scanThread(s.pool.QueryRow(ctx, `SELECT `+threadCols+` FROM threads WHERE id = $1`, id)) + if errors.Is(err, pgx.ErrNoRows) { + return Thread{}, ErrNotFound + } + return t, err +} diff --git a/internal/mail/backend.go b/internal/mail/backend.go new file mode 100644 index 0000000..6c2d51c --- /dev/null +++ b/internal/mail/backend.go @@ -0,0 +1,77 @@ +// Package mail defines the MailBackend abstraction (ARCHITECTURE.md §0.2): all +// inbound delivery and outbound sending sit behind one interface so the core +// (API, MCP, store, threading) never depends on *how* mail moves. Concrete +// backends — relay, imap_smtp, embedded — live in subpackages and are added in +// later milestones. Milestone 1 ships only NullBackend. +package mail + +import ( + "context" + "errors" +) + +// OutgoingMessage is a message the core wants sent. The backend is responsible +// for building/serializing and (where applicable) DKIM-signing it. +type OutgoingMessage struct { + InboxID string + From string + To []string + Cc []string + Bcc []string + Subject string + Text string + HTML string + InReplyTo string + References []string +} + +// SendResult reports the outcome of a Send. +type SendResult struct { + MessageIDHdr string // RFC 5322 Message-ID assigned to the sent message + Accepted bool +} + +// InboundSink receives raw RFC 5322 messages a backend has accepted for an +// address. The core implements this (parse → thread → store → events). +type InboundSink interface { + Deliver(ctx context.Context, inboxAddr string, raw []byte) error +} + +// Caps advertises what a backend can do, so the API/MCP can expose accurate +// capabilities (e.g. whether throwaway addresses or custom domains are possible). +type Caps struct { + SelfHost bool // runs its own MTA in-process + InboundPush bool // delivers inbound without polling (webhook or :25) + CustomDomain bool // can own an arbitrary domain + ThrowawayAddrs bool // can mint addresses on demand +} + +// MailBackend abstracts where mail comes from and how it leaves. +type MailBackend interface { + // Send dispatches an outgoing message. + Send(ctx context.Context, msg *OutgoingMessage) (SendResult, error) + // Start delivers inbound messages to sink until ctx is cancelled. + Start(ctx context.Context, sink InboundSink) error + // Capabilities describes what this backend supports. + Capabilities() Caps +} + +// ErrNotSupported is returned by backends for operations they cannot perform. +var ErrNotSupported = errors.New("mail: operation not supported by this backend") + +// NullBackend satisfies MailBackend without moving any mail. It is the +// milestone-1 default: messages enter only via the ingest API (core acts as its +// own InboundSink), and sending is unavailable until a real backend is wired. +type NullBackend struct{} + +func (NullBackend) Send(context.Context, *OutgoingMessage) (SendResult, error) { + return SendResult{}, ErrNotSupported +} + +// Start blocks until cancelled; the null backend never produces inbound mail. +func (NullBackend) Start(ctx context.Context, _ InboundSink) error { + <-ctx.Done() + return ctx.Err() +} + +func (NullBackend) Capabilities() Caps { return Caps{} }