karti-ai
428040d964
Blockers in the ingest path (all funnel real mail in milestone 2): - H1: coerce body bytes to valid UTF-8 (ToValidUTF8) — non-UTF-8/mid-rune cuts no longer abort the ingest tx and drop the message. - H2: EnsurePart's recoverable parse error is non-fatal — proceed with the guaranteed-usable Part so messy real-world mail is stored, not rejected. - H3: extractBodies descends into message/rfc822 (Part.Message via SetMessageReaderAt) — forwarded/bounce bodies no longer lost. - H4: GetMessage/GetThread/GetThreadMessages scoped to inbox_id — no cross-inbox access; reply no longer a confused deputy. Hardening: - M1: /healthz no longer leaks DB error to unauthenticated callers. - M2: all DB errors funnel through handleErr; malformed UUID -> 404, dup -> 409, internal errors no longer echo the driver string. - M3: index messages(inbox_id, message_id_hdr) for thread resolution. - M4: pods UNIQUE(name) + ON CONFLICT (name) — no duplicate default pods. - L1: skip empty-User/Host addresses (no literal "@"). - L2: skip attachment-disposition parts when picking the body. - L3: case-insensitive, trimmed 'Re:' detection. Verified e2e vs Postgres 16: latin1 body stored valid UTF-8; rfc822-only body extracted; cross-inbox 404; malformed UUID 404; dup 409; threading regression OK. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
58 lines
1.6 KiB
Go
58 lines
1.6 KiB
Go
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()
|
|
}
|
|
|
|
// GetThread is scoped to the inbox: a thread id belonging to another inbox
|
|
// returns ErrNotFound.
|
|
func (s *Service) GetThread(ctx context.Context, inboxID, id string) (Thread, error) {
|
|
t, err := scanThread(s.pool.QueryRow(ctx,
|
|
`SELECT `+threadCols+` FROM threads WHERE id = $1 AND inbox_id = $2`, id, inboxID))
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return Thread{}, ErrNotFound
|
|
}
|
|
return t, err
|
|
}
|