karti-ai
551739baa3
- Confirmed core bet via spike/mimecheck: mox/message parses a messy multipart message standalone (envelope, MIME tree, attachment, body) with only an io.ReaderAt — no mox store/config. Option B is viable. - cmd/openmail single binary: serve|migrate|smtpd|sender|version subcommands. - internal/api: chi router, constant-time bearer auth, /healthz, stubbed v0 AgentMail-shaped routes (501 until core services land). - internal/store: pgxpool + embedded, idempotent, tracked SQL migrations. - internal/store/migrations/0001_init.sql: full native schema (pods, inboxes, threads, messages, attachments, drafts, api_keys, webhooks, outbox, events, domains) with FTS + GIN indexes. Validated end-to-end against Postgres 16. - deploy/: docker-compose (postgres+minio+openmail+caddy), Caddyfile, DNS.md (MX/SPF/DKIM/DMARC/DANE/MTA-STS). Makefile, .gitignore. - Verified: go vet clean; build OK; health/auth smoke tests; migrate idempotent. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
101 lines
2.7 KiB
Go
101 lines
2.7 KiB
Go
// Package store owns OpenMail's Postgres-backed persistence. This is the
|
|
// deliberate divergence from Mox: OpenMail keeps its own native, agent-shaped
|
|
// data model (see ARCHITECTURE.md §3) rather than embedding Mox's bstore/bbolt
|
|
// account store.
|
|
package store
|
|
|
|
import (
|
|
"context"
|
|
"embed"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
//go:embed migrations/*.sql
|
|
var migrationsFS embed.FS
|
|
|
|
type Store struct {
|
|
Pool *pgxpool.Pool
|
|
}
|
|
|
|
// Open connects to Postgres and verifies the connection.
|
|
func Open(ctx context.Context, databaseURL string) (*Store, error) {
|
|
if databaseURL == "" {
|
|
return nil, fmt.Errorf("store: DATABASE_URL is empty")
|
|
}
|
|
pool, err := pgxpool.New(ctx, databaseURL)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("store: connect: %w", err)
|
|
}
|
|
if err := pool.Ping(ctx); err != nil {
|
|
pool.Close()
|
|
return nil, fmt.Errorf("store: ping: %w", err)
|
|
}
|
|
return &Store{Pool: pool}, nil
|
|
}
|
|
|
|
func (s *Store) Close() {
|
|
if s.Pool != nil {
|
|
s.Pool.Close()
|
|
}
|
|
}
|
|
|
|
// Migrate applies any embedded migrations not yet recorded in schema_migrations,
|
|
// in filename order. Each migration runs in its own transaction.
|
|
func (s *Store) Migrate(ctx context.Context) error {
|
|
_, err := s.Pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
version text PRIMARY KEY,
|
|
applied_at timestamptz NOT NULL DEFAULT now()
|
|
)`)
|
|
if err != nil {
|
|
return fmt.Errorf("migrate: ensure schema_migrations: %w", err)
|
|
}
|
|
|
|
entries, err := migrationsFS.ReadDir("migrations")
|
|
if err != nil {
|
|
return fmt.Errorf("migrate: read embedded migrations: %w", err)
|
|
}
|
|
names := make([]string, 0, len(entries))
|
|
for _, e := range entries {
|
|
if strings.HasSuffix(e.Name(), ".sql") {
|
|
names = append(names, e.Name())
|
|
}
|
|
}
|
|
sort.Strings(names)
|
|
|
|
for _, name := range names {
|
|
var exists bool
|
|
if err := s.Pool.QueryRow(ctx,
|
|
`SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version=$1)`, name,
|
|
).Scan(&exists); err != nil {
|
|
return fmt.Errorf("migrate: check %s: %w", name, err)
|
|
}
|
|
if exists {
|
|
continue
|
|
}
|
|
sqlBytes, err := migrationsFS.ReadFile("migrations/" + name)
|
|
if err != nil {
|
|
return fmt.Errorf("migrate: read %s: %w", name, err)
|
|
}
|
|
tx, err := s.Pool.Begin(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("migrate: begin %s: %w", name, err)
|
|
}
|
|
if _, err := tx.Exec(ctx, string(sqlBytes)); err != nil {
|
|
_ = tx.Rollback(ctx)
|
|
return fmt.Errorf("migrate: apply %s: %w", name, err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO schema_migrations(version) VALUES($1)`, name); err != nil {
|
|
_ = tx.Rollback(ctx)
|
|
return fmt.Errorf("migrate: record %s: %w", name, err)
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return fmt.Errorf("migrate: commit %s: %w", name, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|