Milestone 1: core services + Mox ingest + wired v0 API
- 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) <noreply@anthropic.com>
This commit is contained in:
karti-ai
@@ -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
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user