Keep a mail entry per mail, never the body (issue 1517)

This commit is contained in:
jochen
2026-10-20 23:53:48 +02:00
parent d17b7e832d
commit 0d2d19daa0
12 changed files with 967 additions and 12 deletions
@@ -6,15 +6,19 @@
// mail_reader_health (1519 — not yet built). It reads mail; it moves, flags, deletes and sends nothing
// (novox/hq ADR 0459 §11).
//
// mesh-mail-reader the bundle the node's runtime launches: the six mail-reading verbs plus
// mail_reader_health, served over stdio (tools.go)
// mesh-mail-reader the bundle the node's runtime launches: the six mail-reading verbs plus
// mail_reader_health, served over stdio (tools.go)
// mesh-mail-reader prepare the preparation step: the index's schema (store.go, pgstore.go) brought to
// this version's (issue 1517)
//
// stdout is the bundle's MCP channel; what this module says, it says on stderr.
package main
import (
"context"
"fmt"
"os"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
@@ -29,9 +33,36 @@ func envFromOS() Env {
}
func main() {
if len(os.Args) > 1 && os.Args[1] == "prepare" {
os.Exit(prepare())
}
b := &Bundle{Env: envFromOS()}
if err := stdio.ServeOn("", b.Tools(), os.Stdin, os.Stdout); err != nil {
logf("[mesh-mail-reader] %v", err)
os.Exit(1)
}
}
// ---- prepare ------------------------------------------------------------------------------------------------------
// prepare is the preparation step (1517/AC2): the schema brought to what this version needs, run once
// before the bundle itself starts (the manifest's run-once process `prepare`).
func prepare() int {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
s, err := PgStoreFromEnv(ctx)
if err != nil {
logf("[mesh-mail-reader] preparing the index: %v", err)
return 1
}
defer s.Close()
said, err := s.Prepare(ctx)
for _, line := range said {
logf("[mesh-mail-reader] %s", line)
}
if err != nil {
logf("[mesh-mail-reader] preparing the index: %v", err)
return 1
}
return 0
}
@@ -19,14 +19,17 @@ type seat struct {
}
type manifest struct {
Module string `json:"module"`
RunsAs string `json:"runs-as"`
Claims []seat `json:"claims"`
Invokes []string `json:"invokes"`
Tools []string `json:"tools"`
Module string `json:"module"`
RunsAs string `json:"runs-as"`
Claims []seat `json:"claims"`
Invokes []string `json:"invokes"`
Tools []string `json:"tools"`
Requires []string `json:"requires"`
Contributes map[string]map[string]any `json:"contributes"`
Resources []map[string]any `json:"resources"`
}
func readManifest(t *testing.T) manifest {
func readManifest(t *testing.T) (manifest, map[string]map[string]any) {
t.Helper()
raw, err := os.ReadFile(filepath.Join("..", "..", "module.json"))
if err != nil {
@@ -36,14 +39,18 @@ func readManifest(t *testing.T) manifest {
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatal(err)
}
return m
byID := map[string]map[string]any{}
for _, r := range m.Resources {
byID[r["id"].(string)] = r
}
return m, byID
}
// 1426/AC2: runs-as mesh-mail-reader, claims mail-reading (scope mesh) with the six verbs, and invokes
// seat:mail-store.read and seat:mail-store.accounts but no send, send-status or any verb that changes the
// mailbox.
func TestTheManifestClaimsMailReadingAndInvokesReadOnly(t *testing.T) {
m := readManifest(t)
m, _ := readManifest(t)
if m.Module != Module {
t.Fatalf("module %q, want %q", m.Module, Module)
}
@@ -71,7 +78,7 @@ func TestTheManifestClaimsMailReadingAndInvokesReadOnly(t *testing.T) {
// 1426/AC3 (the manifest's own half): the manifest's tools match the bundle's Tools(), in the same order.
func TestTheManifestListsTheSameToolsAsTheBundle(t *testing.T) {
m := readManifest(t)
m, _ := readManifest(t)
var names []string
for _, tool := range (&Bundle{}).Tools() {
names = append(names, tool.Name)
@@ -80,3 +87,31 @@ func TestTheManifestListsTheSameToolsAsTheBundle(t *testing.T) {
t.Fatalf("manifest tools %v, the code serves %v", m.Tools, names)
}
}
// 1517/AC2: the manifest requires postgres-database, binds it, renders database-url and runs a prepare
// process (run-once) with DATABASE_URL_FILE, as mesh-mail does.
func TestTheManifestRequiresItsOwnDatabase(t *testing.T) {
m, byID := readManifest(t)
if !reflect.DeepEqual(m.Requires, []string{"postgres-database"}) {
t.Fatalf("requires %v, want [postgres-database]", m.Requires)
}
if m.Contributes["postgres-database"]["name"] != "mail-reader" {
t.Fatalf("contributes postgres-database %v", m.Contributes["postgres-database"])
}
databaseURL := byID["database-url"]
if databaseURL == nil || databaseURL["type"] != "file" || databaseURL["template"] != "files/database-url" {
t.Fatalf("database-url resource: %v", databaseURL)
}
prepare := byID["prepare"]
if prepare == nil || prepare["type"] != "process" || prepare["run-once"] != true {
t.Fatalf("prepare resource: %v", prepare)
}
run, _ := prepare["run"].([]any)
if len(run) != 2 || run[1] != "prepare" {
t.Fatalf("prepare's run: %v", prepare["run"])
}
env, _ := prepare["env"].(map[string]any)
if env["DATABASE_URL_FILE"] != databaseURL["path"] {
t.Fatalf("prepare's DATABASE_URL_FILE is %v, database-url's own path is %v", env["DATABASE_URL_FILE"], databaseURL["path"])
}
}
@@ -0,0 +1,63 @@
-- mesh-mail-reader's own index (novox/hq ADR 0459 §4, issue 1517): a mail entry per mail group's judged mail
-- (never a body), the mail groups a batch makes of them, the reading lessons a correction leaves, the mails
-- heard but not yet handed to reading, and the reader's own health. No column here holds a mail body or text
-- taken from one.
create table mail_group (
id bigserial primary key,
label text not null default '',
created timestamptz not null default now());
create table mail_entry (
id bigserial primary key,
account text not null,
message_id text not null default '',
folder text not null,
date timestamptz not null,
sender_name text not null default '',
sender_address text not null default '',
subject text not null default '',
mail_group bigint references mail_group (id),
summary text not null default '' check (char_length(summary) <= 160),
mark text check (mark is null or mark in ('matters', 'scam warning', 'neither')),
why text not null default '',
due_date date,
told boolean not null default false);
create index mail_entry_account_date on mail_entry (account, date desc);
create table reading_lesson (
id bigserial primary key,
account text not null default '',
headers jsonb not null default '{}',
change text not null default '',
made timestamptz not null default now(),
undone boolean not null default false);
-- Heard but not yet read (novox/hq ADR 0459 §3, §12, issue 1518): headers only, kept until reading takes them,
-- in the order they were heard (id, a bigserial, orders arrival regardless of the mail's own date). A mail
-- heard twice is kept once: the unique key below is its account, folder, UIDVALIDITY and UID, the same
-- identity mesh-mail's own event carries for a message whose message id may be empty.
create table waiting_mail (
id bigserial primary key,
account text not null,
folder text not null,
message_id text not null default '',
uidvalidity bigint not null default 0,
uid bigint not null default 0,
date timestamptz not null,
sender jsonb not null default '[]',
subject text not null default '',
flags text[] not null default '{}',
size bigint not null default 0,
heard timestamptz not null default now(),
unique (account, folder, uidvalidity, uid));
-- The reader's own health (novox/hq ADR 0459 §12, issue 1519), kept across a restart: one row, singleton by
-- its fixed id. cause is "" while healthy, else one of refused, timed out, out of shape; since is the first
-- failure's own time, not the latest one.
create table reader_health (
id smallint primary key default 1,
healthy boolean not null default true,
cause text not null default '',
why text not null default '',
since timestamptz,
constraint reader_health_singleton check (id = 1));
@@ -0,0 +1,336 @@
package main
// The index on the mesh's postgres (novox/hq ADR 0459 §4, issue 1517): the database the mesh provisions for
// this module through the postgres-database provision, its address in the file DATABASE_URL_FILE names. Its
// schema is numbered migrations in migrations/, applied by the preparation step (`mesh-mail-reader prepare`)
// and recorded in the database itself; a fresh database is given the consolidated schema, schema.sql, at once,
// with every migration recorded as applied. Nothing else changes the schema. Copied from mesh-mail's own
// pgstore.go (Migrations, PgStoreFromEnv, OpenPgStore, Prepare, Ready are schema-agnostic and unchanged in
// shape; only the module's own tables differ, kept in store.go's Store interface).
import (
"context"
"crypto/sha256"
"embed"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io/fs"
"os"
"path"
"regexp"
"sort"
"strconv"
"strings"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgxpool"
)
//go:embed migrations/*.sql
var migrationFiles embed.FS
//go:embed schema.sql
var consolidated string
// Migration is one numbered change of the schema.
type Migration struct {
Number int
Name string
SQL string
Checksum string
}
var migrationFile = regexp.MustCompile(`^(\d{4})-([a-z0-9-]+)\.sql$`)
// Migrations are the migrations this binary carries, in order, numbered from 0001 without a gap (1517/AC1).
func Migrations() ([]Migration, error) { return migrationsFrom(migrationFiles, "migrations") }
// migrationsFrom is Migrations with its directory factored out, so a test can hand it a fake one (fstest.MapFS)
// without touching the real, embedded migrations/ (1517/AC1: a gap or a bad name is refused).
func migrationsFrom(dir fs.FS, name string) ([]Migration, error) {
entries, err := fs.ReadDir(dir, name)
if err != nil {
return nil, err
}
var out []Migration
for _, e := range entries {
m := migrationFile.FindStringSubmatch(e.Name())
if m == nil {
return nil, fmt.Errorf("%s is not a migration's name (NNNN-what-it-does.sql)", e.Name())
}
n, _ := strconv.Atoi(m[1])
body, err := fs.ReadFile(dir, path.Join(name, e.Name()))
if err != nil {
return nil, err
}
if strings.TrimSpace(string(body)) == "" {
return nil, fmt.Errorf("%s is empty", e.Name())
}
sum := sha256.Sum256(body)
out = append(out, Migration{Number: n, Name: m[2], SQL: string(body), Checksum: hex.EncodeToString(sum[:])})
}
sort.Slice(out, func(i, j int) bool { return out[i].Number < out[j].Number })
for i, m := range out {
if m.Number != i+1 {
return nil, fmt.Errorf("the migrations are not numbered 0001 on without a gap: %04d is the %d. one", m.Number, i+1)
}
}
return out, nil
}
const ledger = `create table if not exists migration (
number integer primary key,
name text not null,
checksum text not null,
applied timestamptz not null default now())`
// lockKey is the advisory lock every migration run takes, so two preparations never interleave. Its own value,
// distinct from mesh-mail's (4_250_359_001), so the two modules' preparations never contend for the same lock
// even if they ever shared a database (they do not: each requires its own postgres-database).
const lockKey int64 = 4_250_459_001
// PgStore is the Store (store.go) on postgres.
type PgStore struct{ pool *pgxpool.Pool }
// PgStoreFromEnv opens the index the mesh provisioned.
func PgStoreFromEnv(ctx context.Context) (*PgStore, error) {
file := os.Getenv("DATABASE_URL_FILE")
if file == "" {
return nil, errors.New("DATABASE_URL_FILE is not set: the index's database is a requirement the mesh resolves")
}
raw, err := os.ReadFile(file)
if err != nil {
return nil, fmt.Errorf("the database's address cannot be read: %w", err)
}
return OpenPgStore(ctx, strings.TrimSpace(string(raw)))
}
// OpenPgStore opens the index at an address.
func OpenPgStore(ctx context.Context, url string) (*PgStore, error) {
pool, err := pgxpool.New(ctx, url)
if err != nil {
return nil, err
}
return &PgStore{pool: pool}, nil
}
// Close closes the store.
func (s *PgStore) Close() { s.pool.Close() }
// Prepare brings the schema to what this version needs, and says what it did (1517/AC2).
func (s *PgStore) Prepare(ctx context.Context) ([]string, error) {
migrations, err := Migrations()
if err != nil {
return nil, err
}
conn, err := s.pool.Acquire(ctx)
if err != nil {
return nil, err
}
defer conn.Release()
if _, err := conn.Exec(ctx, `select pg_advisory_lock($1)`, lockKey); err != nil {
return nil, fmt.Errorf("the migration lock cannot be taken: %w", err)
}
defer func() { _, _ = conn.Exec(context.WithoutCancel(ctx), `select pg_advisory_unlock($1)`, lockKey) }()
if _, err := conn.Exec(ctx, ledger); err != nil {
return nil, err
}
applied := map[int]string{}
rows, err := conn.Query(ctx, `select number, checksum from migration`)
if err != nil {
return nil, err
}
highest := 0
for rows.Next() {
var n int
var sum string
if err := rows.Scan(&n, &sum); err != nil {
rows.Close()
return nil, err
}
applied[n] = sum
highest = max(highest, n)
}
rows.Close()
if err := rows.Err(); err != nil {
return nil, err
}
record := func(tx pgx.Tx, m Migration) error {
_, err := tx.Exec(ctx, `insert into migration (number, name, checksum) values ($1, $2, $3)`, m.Number, m.Name, m.Checksum)
return err
}
if len(applied) == 0 {
var exists bool
if err := conn.QueryRow(ctx, `select to_regclass('mail_entry') is not null`).Scan(&exists); err != nil {
return nil, err
}
if exists {
return nil, errors.New("the database holds the reader's tables and no record of a migration: its contents are not known, so nothing is changed")
}
tx, err := conn.Begin(ctx)
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
if _, err := tx.Exec(ctx, consolidated); err != nil {
return nil, fmt.Errorf("the consolidated schema failed, and nothing of it was kept: %w", err)
}
for _, m := range migrations {
if err := record(tx, m); err != nil {
return nil, err
}
}
if err := tx.Commit(ctx); err != nil {
return nil, err
}
return []string{fmt.Sprintf("a fresh database: the consolidated schema, through migration %04d", len(migrations))}, nil
}
var problems []string
var pending []Migration
for _, m := range migrations {
sum, ran := applied[m.Number]
switch {
case ran && sum != m.Checksum:
problems = append(problems, fmt.Sprintf("migration %04d-%s ran, and its file has changed since: write a new one", m.Number, m.Name))
case !ran && m.Number < highest:
problems = append(problems, fmt.Sprintf("migration %04d-%s never ran, and %04d did: renumber it above %04d", m.Number, m.Name, highest, highest))
case !ran:
pending = append(pending, m)
}
}
if len(problems) > 0 {
return nil, errors.New("this database and these migrations disagree: " + strings.Join(problems, "; "))
}
var said []string
for _, m := range pending {
err := func() error {
tx, err := conn.Begin(ctx)
if err != nil {
return err
}
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
if _, err := tx.Exec(ctx, m.SQL); err != nil {
return fmt.Errorf("migration %04d-%s failed, and nothing it did was kept: %w", m.Number, m.Name, err)
}
if err := record(tx, m); err != nil {
return err
}
return tx.Commit(ctx)
}()
if err != nil {
return said, err
}
said = append(said, fmt.Sprintf("migration %04d-%s applied", m.Number, m.Name))
}
if len(said) == 0 {
said = []string{fmt.Sprintf("the schema is at migration %04d, as this version needs", len(migrations))}
}
return said, nil
}
// Ready says why the index cannot serve this version, or nil.
func (s *PgStore) Ready(ctx context.Context) error {
migrations, err := Migrations()
if err != nil {
return err
}
var at int
err = s.pool.QueryRow(ctx, `select coalesce(max(number), 0) from migration`).Scan(&at)
var pgErr *pgconn.PgError
if errors.As(err, &pgErr) && pgErr.Code == "42P01" {
at = 0
} else if err != nil {
return fmt.Errorf("the database cannot be reached: %w", err)
}
if at != len(migrations) {
return fmt.Errorf("the database's schema is at migration %04d and this version needs %04d: the preparation step "+
"(mesh-mail-reader prepare) has not run", at, len(migrations))
}
return nil
}
// ---- Store (store.go) over postgres ---------------------------------------------------------------------------
func (s *PgStore) AddWaiting(ctx context.Context, w WaitingMail) error {
from, _ := json.Marshal(orNonePersons(w.From))
flags := w.Flags
if flags == nil {
flags = []string{}
}
_, err := s.pool.Exec(ctx, `insert into waiting_mail (account, folder, message_id, uidvalidity, uid, date, sender, subject, flags, size)
values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
on conflict (account, folder, uidvalidity, uid) do nothing`,
w.Account, w.Folder, w.MessageID, int64(w.UIDValidity), int64(w.UID), w.Date, from, w.Subject, flags, w.Size)
return err
}
func orNonePersons(ps []Person) []Person {
if ps == nil {
return []Person{}
}
return ps
}
func (s *PgStore) Waiting(ctx context.Context) ([]WaitingMail, error) {
rows, err := s.pool.Query(ctx, `select id, account, folder, message_id, uidvalidity, uid, date, sender, subject, flags, size, heard
from waiting_mail order by id asc`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []WaitingMail
for rows.Next() {
var w WaitingMail
var uidvalidity, uid int64
var from []byte
if err := rows.Scan(&w.Seq, &w.Account, &w.Folder, &w.MessageID, &uidvalidity, &uid, &w.Date, &from, &w.Subject, &w.Flags, &w.Size, &w.Heard); err != nil {
return nil, err
}
w.UIDValidity, w.UID = uint32(uidvalidity), uint32(uid)
if err := json.Unmarshal(from, &w.From); err != nil {
return nil, err
}
w.Date, w.Heard = w.Date.UTC(), w.Heard.UTC()
out = append(out, w)
}
return out, rows.Err()
}
func (s *PgStore) TakeWaiting(ctx context.Context, seq int64) error {
_, err := s.pool.Exec(ctx, `delete from waiting_mail where id = $1`, seq)
return err
}
func (s *PgStore) ReaderHealth(ctx context.Context) (ReaderHealth, error) {
var h ReaderHealth
var cause, why string
var since *time.Time
err := s.pool.QueryRow(ctx, `select healthy, cause, why, since from reader_health where id = 1`).Scan(&h.Healthy, &cause, &why, &since)
if errors.Is(err, pgx.ErrNoRows) {
return ReaderHealth{Healthy: true}, nil
}
if err != nil {
return ReaderHealth{}, err
}
h.Cause, h.Why = HealthOutcome(cause), why
if since != nil {
h.Since = since.UTC()
}
return h, nil
}
func (s *PgStore) SetReaderHealth(ctx context.Context, h ReaderHealth) error {
var since *time.Time
if !h.Since.IsZero() {
t := h.Since
since = &t
}
_, err := s.pool.Exec(ctx, `insert into reader_health (id, healthy, cause, why, since) values (1, $1, $2, $3, $4)
on conflict (id) do update set healthy = excluded.healthy, cause = excluded.cause, why = excluded.why, since = excluded.since`,
h.Healthy, string(h.Cause), h.Why, since)
return err
}
@@ -0,0 +1,139 @@
//go:build integration
package main
// 1517/AC4: against a real postgres (MESH_TEST_POSTGRES), prepare on an empty database applies the
// consolidated schema and records every migration; on a database at migration 0001 it applies what follows;
// both end with the same tables. This is compiled (go vet -tags integration ./...) but never run from here:
// the lead's mesh-controller.try runs it on the pushed branch (the context file for this milestone, and
// novox/hq's own rule for integration tests). It fails, never skips, without MESH_TEST_POSTGRES.
import (
"context"
"net/url"
"os"
"reflect"
"strings"
"testing"
"github.com/jackc/pgx/v5/pgxpool"
)
func testDatabaseURL(t *testing.T) string {
t.Helper()
url := os.Getenv("MESH_TEST_POSTGRES")
if url == "" {
t.Fatal("MESH_TEST_POSTGRES is unset: this integration test needs a real postgres, and never skips")
}
return url
}
// freshIntegrationDatabase makes a new, empty database on the server MESH_TEST_POSTGRES names, and returns its
// own URL.
func freshIntegrationDatabase(t *testing.T) string {
t.Helper()
ctx := context.Background()
base := testDatabaseURL(t)
admin, err := pgxpool.New(ctx, base)
if err != nil {
t.Fatal(err)
}
defer admin.Close()
name := "mesh_mail_reader_test_" + strings.ReplaceAll(t.Name(), "/", "_")
name = strings.ToLower(name)
if _, err := admin.Exec(ctx, "drop database if exists "+name); err != nil {
t.Fatal(err)
}
if _, err := admin.Exec(ctx, "create database "+name); err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
cleanup, err := pgxpool.New(context.Background(), base)
if err == nil {
_, _ = cleanup.Exec(context.Background(), "drop database if exists "+name)
cleanup.Close()
}
})
u, err := url.Parse(base)
if err != nil {
t.Fatal(err)
}
u.Path = "/" + name
return u.String()
}
// 1517/AC4.
func TestPrepareOnAFreshDatabaseAppliesTheConsolidatedSchema(t *testing.T) {
ctx := context.Background()
fresh, err := OpenPgStore(ctx, freshIntegrationDatabase(t))
if err != nil {
t.Fatal(err)
}
defer fresh.Close()
said, err := fresh.Prepare(ctx)
if err != nil {
t.Fatal(err)
}
if len(said) == 0 || !strings.Contains(said[0], "a fresh database") {
t.Fatalf("said %v, want it to say it gave a fresh database the consolidated schema", said)
}
if err := fresh.Ready(ctx); err != nil {
t.Fatal(err)
}
}
// 1517/AC4: a database already at migration 0001 (the consolidated schema's own migration) applies whatever
// comes after it, and ends at the same tables as a fresh database given the consolidated schema outright.
func TestPrepareOnADatabaseAtMigration0001AppliesWhatFollows(t *testing.T) {
ctx := context.Background()
fresh, err := OpenPgStore(ctx, freshIntegrationDatabase(t))
if err != nil {
t.Fatal(err)
}
defer fresh.Close()
if _, err := fresh.Prepare(ctx); err != nil {
t.Fatal(err)
}
stepped, err := OpenPgStore(ctx, freshIntegrationDatabase(t))
if err != nil {
t.Fatal(err)
}
defer stepped.Close()
migrations, err := Migrations()
if err != nil {
t.Fatal(err)
}
for _, m := range migrations {
if _, err := stepped.pool.Exec(ctx, m.SQL); err != nil {
t.Fatalf("migration %04d: %v", m.Number, err)
}
}
if _, err := stepped.Prepare(ctx); err != nil {
t.Fatal(err)
}
a, b := integrationSchemaShape(t, fresh), integrationSchemaShape(t, stepped)
if !reflect.DeepEqual(a, b) {
t.Fatalf("the consolidated schema and the migrations differ:\nconsolidated %v\nmigrations %v", a, b)
}
}
func integrationSchemaShape(t *testing.T, s *PgStore) []string {
t.Helper()
rows, err := s.pool.Query(context.Background(), `select table_name || '.' || column_name || ' ' || data_type
from information_schema.columns where table_schema = 'public' and table_name <> 'migration' order by 1`)
if err != nil {
t.Fatal(err)
}
defer rows.Close()
var out []string
for rows.Next() {
var c string
if err := rows.Scan(&c); err != nil {
t.Fatal(err)
}
out = append(out, c)
}
return out
}
@@ -0,0 +1,84 @@
package main
// Unit tests for the schema (novox/hq ADR 0459 §4, issue 1517): the migrations are read with no gap and a
// bad name is refused (AC1), and no table holds a mail body or text taken from one (AC3). AC2 (the manifest
// requires postgres-database, binds it, renders database-url and runs prepare) is manifest_test.go's; AC4
// (against a real postgres) is pgstore_integration_test.go, compiled but never run from here.
import (
"strings"
"testing"
"testing/fstest"
)
// 1517/AC1: Migrations() reads migrations/0001-*.sql onward with no gap.
func TestMigrationsReadsTheRealOnesWithNoGap(t *testing.T) {
ms, err := Migrations()
if err != nil {
t.Fatal(err)
}
if len(ms) == 0 {
t.Fatal("no migrations found")
}
for i, m := range ms {
if m.Number != i+1 {
t.Fatalf("migration %d is numbered %04d, want %04d", i, m.Number, i+1)
}
}
}
// 1517/AC1: a gap in the numbering is refused.
func TestMigrationsRefusesAGap(t *testing.T) {
fsys := fstest.MapFS{
"m/0001-first.sql": {Data: []byte("create table a (id int);")},
"m/0003-third.sql": {Data: []byte("create table b (id int);")},
}
if _, err := migrationsFrom(fsys, "m"); err == nil || !strings.Contains(err.Error(), "gap") {
t.Fatalf("migrationsFrom with a gap: %v, want an error naming the gap", err)
}
}
// 1517/AC1: a file whose name is not NNNN-what-it-does.sql is refused by name.
func TestMigrationsRefusesABadName(t *testing.T) {
fsys := fstest.MapFS{
"m/0001-first.sql": {Data: []byte("create table a (id int);")},
"m/not-a-migration.sql": {Data: []byte("create table b (id int);")},
}
if _, err := migrationsFrom(fsys, "m"); err == nil || !strings.Contains(err.Error(), "not a migration's name") {
t.Fatalf("migrationsFrom with a bad name: %v, want an error naming it", err)
}
}
// forbiddenColumn is 1517/AC3: no table in schema.sql or any migration has a column for a mail body or text.
var forbiddenColumns = []string{"body", "text", "content", "snippet"}
// 1517/AC3: no table in schema.sql or any migration has a column named body, text, content or snippet, and
// summary is bounded to 160 characters by a CHECK.
func TestTheSchemaKeepsNoMailBodyOrText(t *testing.T) {
for _, sql := range allSchemaSQL(t) {
for _, line := range strings.Split(sql, "\n") {
word := strings.ToLower(strings.TrimSpace(strings.Fields(line + " x")[0]))
for _, forbidden := range forbiddenColumns {
if word == forbidden {
t.Fatalf("a column is named %q, which may hold a mail body or text taken from one: %q", forbidden, line)
}
}
}
}
if !strings.Contains(consolidated, "char_length(summary) <= 160") {
t.Fatal("summary has no CHECK bounding it to 160 characters")
}
}
func allSchemaSQL(t *testing.T) []string {
t.Helper()
ms, err := Migrations()
if err != nil {
t.Fatal(err)
}
out := []string{consolidated}
for _, m := range ms {
out = append(out, m.SQL)
}
return out
}
@@ -0,0 +1,61 @@
-- The consolidated schema, through migration 0001: what a fresh database is given at once (the mesh's rule: a
-- numbered migration for the running store, and this for a fresh install).
create table mail_group (
id bigserial primary key,
label text not null default '',
created timestamptz not null default now());
create table mail_entry (
id bigserial primary key,
account text not null,
message_id text not null default '',
folder text not null,
date timestamptz not null,
sender_name text not null default '',
sender_address text not null default '',
subject text not null default '',
mail_group bigint references mail_group (id),
summary text not null default '' check (char_length(summary) <= 160),
mark text check (mark is null or mark in ('matters', 'scam warning', 'neither')),
why text not null default '',
due_date date,
told boolean not null default false);
create index mail_entry_account_date on mail_entry (account, date desc);
create table reading_lesson (
id bigserial primary key,
account text not null default '',
headers jsonb not null default '{}',
change text not null default '',
made timestamptz not null default now(),
undone boolean not null default false);
-- Heard but not yet read (novox/hq ADR 0459 §3, §12, issue 1518): headers only, kept until reading takes them,
-- in the order they were heard (id, a bigserial, orders arrival regardless of the mail's own date). A mail
-- heard twice is kept once: the unique key below is its account, folder, UIDVALIDITY and UID, the same
-- identity mesh-mail's own event carries for a message whose message id may be empty.
create table waiting_mail (
id bigserial primary key,
account text not null,
folder text not null,
message_id text not null default '',
uidvalidity bigint not null default 0,
uid bigint not null default 0,
date timestamptz not null,
sender jsonb not null default '[]',
subject text not null default '',
flags text[] not null default '{}',
size bigint not null default 0,
heard timestamptz not null default now(),
unique (account, folder, uidvalidity, uid));
-- The reader's own health (novox/hq ADR 0459 §12, issue 1519), kept across a restart: one row, singleton by
-- its fixed id. cause is "" while healthy, else one of refused, timed out, out of shape; since is the first
-- failure's own time, not the latest one.
create table reader_health (
id smallint primary key default 1,
healthy boolean not null default true,
cause text not null default '',
why text not null default '',
since timestamptz,
constraint reader_health_singleton check (id = 1));
@@ -0,0 +1,130 @@
package main
// The reader's own index (novox/hq ADR 0459 §4): mail entries, mail groups and reading lessons for the verbs
// (1521, 1428), the waiting mails heard but not yet read (1518) and the reader's own health state (1519), each
// over postgres (pgstore.go) or in memory for the tests. Nothing here keeps a mail body or text taken from one.
import (
"context"
"time"
)
// Person is a sender, as mesh-mail's own event names one (name and address, never a body).
type Person struct {
Name string `json:"name,omitempty"`
Address string `json:"address"`
}
// Mark is an entry's mark (novox/hq ADR 0459 §7).
type Mark string
const (
MarkMatters Mark = "matters"
MarkScamWarning Mark = "scam warning"
MarkNeither Mark = "neither"
)
// Entry is one mail entry (ADR 0459 §4): the mail's own headers (never its body), the mail group it was
// batched into, a one-line summary, a mark and why, a due date, and whether it was told on the mail signal.
type Entry struct {
ID int64
Account string
MessageID string
Folder string
Date time.Time
SenderName string
SenderAddress string
Subject string
Group int64 // the mail group's id, 0 when ungrouped
Summary string
Mark Mark // "" until a batch judges it
Why string
DueDate time.Time // zero when none
Told bool
}
// Group is one mail group (ADR 0459 §4).
type Group struct {
ID int64
Label string
Created time.Time
}
// Lesson is one reading lesson (ADR 0459 §8, §9): the mail's own headers, the change made, when, and whether
// it was undone (1523).
type Lesson struct {
ID int64
Account string
Headers map[string]any
Change string
Made time.Time
Undone bool
}
// WaitingMail is one mail heard (an added event) but not yet handed to reading (ADR 0459 §3, §12, issue 1518):
// headers only, never a body. Seq is the order it was heard in — a monotonic sequence, independent of Date —
// so catch-up after an outage reads in arrival order (1518/AC4), including mail heard while reading was
// unhealthy.
type WaitingMail struct {
Seq int64
Account string
Folder string
MessageID string
UIDValidity uint32
UID uint32
Date time.Time
From []Person
Subject string
Flags []string
Size int64
Heard time.Time
}
// Key is the waiting mail's own identity: its message id wherever it has one (stable across a UIDVALIDITY
// reset), else its account, folder, UIDVALIDITY and UID (novox/hq ADR 0459 §3; the same fallback mesh-mail's
// own event dedupe, mesh-mail-notifier's EventBody.Handle, uses).
func (w WaitingMail) Key() string {
if w.MessageID != "" {
return w.Account + "\x1f" + w.MessageID
}
return w.Account + "\x1f" + w.Folder
}
// HealthOutcome names why a batch's judgement did not succeed (novox/hq ADR 0459 §12); the empty outcome means
// it succeeded.
type HealthOutcome string
const (
OutcomeOK HealthOutcome = ""
OutcomeRefused HealthOutcome = "refused"
OutcomeTimedOut HealthOutcome = "timed out"
OutcomeOutOfShape HealthOutcome = "out of shape"
)
// ReaderHealth is the kept health state (1519): unhealthy once a judgement is refused, times out or answers out
// of shape, naming the cause, the judgement's own words, and since when; healthy again once one succeeds. Kept
// in the module's own database, so a restart answers the same (1519/AC4).
type ReaderHealth struct {
Healthy bool
Cause HealthOutcome // "" while healthy
Why string // the judgement's own words
Since time.Time // the first failure's own time; zero while healthy
}
// Store is everything mesh-mail-reader keeps of its own. A PgStore (pgstore.go) backs it in the running
// module; the tests use an in-memory double (kit_test.go).
type Store interface {
// AddWaiting records an added event as a waiting mail before the event is acknowledged (1518/AC2): a mail
// heard is never lost, even if the module crashes before it is read. The same mail heard twice (1518/AC3,
// matched by WaitingMail.Key) is recorded once; AddWaiting itself never errors on a duplicate.
AddWaiting(ctx context.Context, w WaitingMail) error
// Waiting answers every waiting mail in arrival order, oldest first (1518/AC4).
Waiting(ctx context.Context) ([]WaitingMail, error)
// TakeWaiting removes a waiting mail once it has been handed to reading (1520); not yet called before 1519.
TakeWaiting(ctx context.Context, seq int64) error
// ReaderHealth answers the kept health state (1519/AC4); healthy with a zero Since on a fresh database.
ReaderHealth(ctx context.Context) (ReaderHealth, error)
// SetReaderHealth replaces the kept health state.
SetReaderHealth(ctx context.Context, h ReaderHealth) error
}
@@ -0,0 +1 @@
postgresql://${bound:postgres-database:as}:${secret:postgres-database}@${bound:postgres-database:at}:${bound:postgres-database:port}/${bound:postgres-database:as}
+12 -1
View File
@@ -2,4 +2,15 @@ module mesh-mail-reader
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019062050-34392e74916c
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019062050-34392e74916c
github.com/jackc/pgx/v5 v5.11.0
)
require (
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
github.com/jackc/puddle/v2 v2.2.2 // indirect
golang.org/x/sync v0.17.0 // indirect
golang.org/x/text v0.29.0 // indirect
)
+26
View File
@@ -1,2 +1,28 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019062050-34392e74916c h1:/f6f2ugss172hDJSL3EFZqb9PdfO5DKzYVfPiX2LdAM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019062050-34392e74916c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
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/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg=
github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
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/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
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=
+38
View File
@@ -13,6 +13,18 @@
"seat:mail-store.read",
"seat:mail-store.accounts"
],
"requires": ["postgres-database"],
"contributes": {
"postgres-database": {
"name": "mail-reader"
}
},
"binds": {
"postgres-database": "${dir:state}/database.json"
},
"secrets": {
"postgres-database": "${dir:state}/database.secret"
},
"config": {
"callers": {
"type": "list",
@@ -60,6 +72,31 @@
"merge": "json",
"template": "files/callers.json",
"takes-config": ["callers"]
},
{
"id": "database-url",
"type": "file",
"path": "${dir:state}/database.url",
"mode": "0600",
"owner": "mesh-mail-reader",
"template": "files/database-url"
},
{
"id": "prepare",
"type": "process",
"name": "mesh-mail-reader-prepare",
"artifact": "tools",
"run": [
"./mesh-mail-reader",
"prepare"
],
"run-once": true,
"env": {
"DATABASE_URL_FILE": "${dir:state}/database.url"
},
"restart-on": [
"database-url"
]
}
],
"build": {
@@ -75,6 +112,7 @@
"mesh-mail-reader"
],
"env": {
"DATABASE_URL_FILE": "${dir:state}/database.url",
"MESH_MAIL_READER_CALLERS": "${dir:state}/callers.json"
}
},