diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go index 3796f6f6..ace6f32a 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go @@ -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 +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go index bf85282d..b94e9d9c 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go @@ -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"]) + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0001-entries-groups-lessons-waiting-mail-and-health-keep-no-body.sql b/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0001-entries-groups-lessons-waiting-mail-and-health-keep-no-body.sql new file mode 100644 index 00000000..f3712f14 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0001-entries-groups-lessons-waiting-mail-and-health-keep-no-body.sql @@ -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)); diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go new file mode 100644 index 00000000..92289b16 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go @@ -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 +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore_integration_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore_integration_test.go new file mode 100644 index 00000000..a85cd8a6 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore_integration_test.go @@ -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 +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore_test.go new file mode 100644 index 00000000..85faf80a --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore_test.go @@ -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 +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql b/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql new file mode 100644 index 00000000..c3383487 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql @@ -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)); diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go new file mode 100644 index 00000000..9dff4b3a --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go @@ -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 +} diff --git a/modules/mesh-mail-reader/files/database-url b/modules/mesh-mail-reader/files/database-url new file mode 100644 index 00000000..dc59da9a --- /dev/null +++ b/modules/mesh-mail-reader/files/database-url @@ -0,0 +1 @@ +postgresql://${bound:postgres-database:as}:${secret:postgres-database}@${bound:postgres-database:at}:${bound:postgres-database:port}/${bound:postgres-database:as} diff --git a/modules/mesh-mail-reader/go.mod b/modules/mesh-mail-reader/go.mod index e9fc7dd3..cd92bafa 100644 --- a/modules/mesh-mail-reader/go.mod +++ b/modules/mesh-mail-reader/go.mod @@ -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 +) diff --git a/modules/mesh-mail-reader/go.sum b/modules/mesh-mail-reader/go.sum index 25aaa963..860456a6 100644 --- a/modules/mesh-mail-reader/go.sum +++ b/modules/mesh-mail-reader/go.sum @@ -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= diff --git a/modules/mesh-mail-reader/module.json b/modules/mesh-mail-reader/module.json index 82ddc7f4..d00618ad 100644 --- a/modules/mesh-mail-reader/module.json +++ b/modules/mesh-mail-reader/module.json @@ -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" } },