diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go index 99693fa4..e738d67a 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go @@ -269,7 +269,7 @@ func RunOneBatch(ctx context.Context, store Store, readText ReadTextFunc, judge } name, address := senderOf(w.From) e := Entry{ - Account: w.Account, MessageID: w.MessageID, Folder: w.Folder, Date: w.Date, + Account: w.Account, MessageID: w.MessageID, Folder: w.Folder, UIDValidity: w.UIDValidity, UID: w.UID, Date: w.Date, SenderName: name, SenderAddress: address, Subject: w.Subject, Group: groupID, Summary: j.Summary, Mark: j.Mark, Why: j.Why, DueDate: due, Judged: judgedAt, } diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/entries_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/entries_test.go index 00b6c071..f7bc81dc 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/entries_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/entries_test.go @@ -111,8 +111,8 @@ func TestEntriesToolAnswersFilteredAndSorted(t *testing.T) { if err != nil { t.Fatal(err) } - store.AddEntry(ctx, Entry{Account: "bob", Subject: "a", Mark: MarkNeither, Group: g.ID, Date: time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)}) - store.AddEntry(ctx, Entry{Account: "bob", Subject: "b", Mark: MarkMatters, DueDate: time.Date(2026, 11, 1, 0, 0, 0, 0, time.UTC), Date: time.Date(2026, 10, 2, 0, 0, 0, 0, time.UTC)}) + store.AddEntry(ctx, Entry{Account: "bob", UID: 1, Subject: "a", Mark: MarkNeither, Group: g.ID, Date: time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)}) + store.AddEntry(ctx, Entry{Account: "bob", UID: 2, Subject: "b", Mark: MarkMatters, DueDate: time.Date(2026, 11, 1, 0, 0, 0, 0, time.UTC), Date: time.Date(2026, 10, 2, 0, 0, 0, 0, time.UTC)}) b := &Bundle{ Caller: func(map[string]any) string { return "mesh-mail-web" }, @@ -145,9 +145,9 @@ func TestGroupsToolAnswersLargestFirst(t *testing.T) { store := newMemStore() small, _ := store.FindOrCreateGroup(ctx, "small") big, _ := store.FindOrCreateGroup(ctx, "big") - store.AddEntry(ctx, Entry{Group: small.ID}) - store.AddEntry(ctx, Entry{Group: big.ID}) - store.AddEntry(ctx, Entry{Group: big.ID}) + store.AddEntry(ctx, Entry{UID: 1, Group: small.ID}) + store.AddEntry(ctx, Entry{UID: 2, Group: big.ID}) + store.AddEntry(ctx, Entry{UID: 3, Group: big.ID}) b := &Bundle{ Caller: func(map[string]any) string { return "mesh-mail-web" }, diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go index 6d831299..53109853 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go @@ -13,7 +13,6 @@ import ( type memStore struct { mu sync.Mutex waiting []WaitingMail - keys map[string]bool nextSeq int64 failNext error // AddWaiting fails once with this, then clears, so a test can prove a failure is returned health ReaderHealth @@ -27,7 +26,7 @@ type memStore struct { } func newMemStore() *memStore { - return &memStore{keys: map[string]bool{}, health: ReaderHealth{Healthy: true}} + return &memStore{health: ReaderHealth{Healthy: true}} } func (s *memStore) AddWaiting(_ context.Context, w WaitingMail) error { @@ -38,11 +37,18 @@ func (s *memStore) AddWaiting(_ context.Context, w WaitingMail) error { s.failNext = nil return err } - key := w.Key() - if s.keys[key] { - return nil // 1518/AC3: the same mail heard twice is recorded once. + // The same identity PgStore's own unique keys hold (issue 1548: this double once kept a key of its own, + // which hid 1849 from every test): a mail waits once, and not at all once it has its entry (1849/AC1). + for _, held := range s.waiting { + if held.Key() == w.Key() { + return nil // 1518/AC3: the same mail heard twice is recorded once. + } + } + for _, e := range s.entries { + if e.Account == w.Account && entryKey(e) == MessageKey(w.MessageID, w.Folder, w.UIDValidity, w.UID) { + return nil + } } - s.keys[key] = true s.nextSeq++ w.Seq = s.nextSeq s.waiting = append(s.waiting, w) @@ -85,12 +91,22 @@ func (s *memStore) SetReaderHealth(_ context.Context, h ReaderHealth) error { func (s *memStore) AddEntry(_ context.Context, e Entry) (int64, error) { s.mu.Lock() defer s.mu.Unlock() + for i, held := range s.entries { + if held.Account == e.Account && entryKey(held) == entryKey(e) { + // As PgStore's on conflict: the judgement replaced, the id, headers and told kept (1849). + e.ID, e.Folder, e.UIDValidity, e.UID, e.Told = held.ID, held.Folder, held.UIDValidity, held.UID, held.Told + s.entries[i] = e + return e.ID, nil + } + } s.nextEntry++ e.ID = s.nextEntry s.entries = append(s.entries, e) return e.ID, nil } +func entryKey(e Entry) string { return MessageKey(e.MessageID, e.Folder, e.UIDValidity, e.UID) } + func (s *memStore) UpdateEntry(_ context.Context, e Entry) error { s.mu.Lock() defer s.mu.Unlock() diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0002-one-entry-per-message.sql b/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0002-one-entry-per-message.sql new file mode 100644 index 00000000..ba7c6ea2 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0002-one-entry-per-message.sql @@ -0,0 +1,22 @@ +-- Issue 1849: one mail entry per message. Gmail files one message in several folders (INBOX and All Mail), +-- each with its own UID, and mesh-mail says an added event per folder; a mail is now kept once per account +-- and message key: its Message-ID ('mid:' and the id) where it has one, else its folder, UIDVALIDITY and UID +-- ('uid:' folder/uidvalidity/uid; MessageKey, store.go). The copies already kept twice are dropped, the +-- oldest kept. +alter table waiting_mail add column message_key text not null default ''; +update waiting_mail set message_key = case when btrim(message_id) <> '' then 'mid:' || btrim(message_id) + else 'uid:' || folder || '/' || uidvalidity || '/' || uid end; +delete from waiting_mail w using waiting_mail o + where w.account = o.account and w.message_key = o.message_key and w.id > o.id; +alter table waiting_mail add constraint waiting_mail_one_per_message unique (account, message_key); + +-- An entry kept before this knows no UID: one without a Message-ID keeps a key of its own ('entry:' and its +-- id), so none of those is merged with another. +alter table mail_entry add column uidvalidity bigint not null default 0; +alter table mail_entry add column uid bigint not null default 0; +alter table mail_entry add column message_key text not null default ''; +update mail_entry set message_key = case when btrim(message_id) <> '' then 'mid:' || btrim(message_id) + else 'entry:' || id end; +delete from mail_entry e using mail_entry o + where e.account = o.account and e.message_key = o.message_key and e.id > o.id; +alter table mail_entry add constraint mail_entry_one_per_message unique (account, message_key); diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/onemail_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/onemail_test.go new file mode 100644 index 00000000..97d84387 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/onemail_test.go @@ -0,0 +1,114 @@ +package main + +// Issue 1849: one mail entry per message. Gmail files one message in several folders (INBOX and All Mail), +// each with its own UID, and mesh-mail says an added event per folder; mail reading keeps one entry per +// message of an account — by its Message-ID where it has one, else by its folder, UIDVALIDITY and UID. + +import ( + "context" + "encoding/json" + "testing" + "time" +) + +// heardIn is mesh-mail's own added event for one message in one folder (metadata(), cmd/mesh-mail/imap.go). +func heardIn(folder string, uid uint32, messageID string) json.RawMessage { + raw, _ := json.Marshal(map[string]any{ + "account": "work", "folder": folder, "uid": uid, "uidvalidity": 7, "message-id": messageID, + "in-reply-to": "", "date": "2026-10-22T09:00:00Z", + "from": []map[string]any{{"name": "Search Console", "address": "sc-noreply@example.com"}}, + "to": []map[string]any{{"address": "op@example.org"}}, + "subject": "New 404 errors", "size": 2048, "flags": []string{}, "attachments": []any{}, + }) + return raw +} + +// readAll runs batches until nothing waits, judging every mail "matters", and counts the mails judged. +func readAll(t *testing.T, store Store) int { + t.Helper() + ctx := context.Background() + judged := 0 + for i := 0; i < 10; i++ { + waiting, err := store.Waiting(ctx) + if err != nil { + t.Fatal(err) + } + if len(waiting) == 0 { + return judged + } + judged += len(waiting) + now := func() time.Time { return time.Date(2026, 10, 22, 10, 0, 0, 0, time.UTC) } + if err := RunOneBatch(ctx, store, noopReadText, judgeAllNeither(waiting), waiting, now, noopLogf); err != nil { + t.Fatal(err) + } + } + t.Fatal("mails still wait after 10 batches") + return judged +} + +// 1849/AC1: the same Gmail message (one Message-ID) heard in INBOX and in All Mail is one entry, judged once +// — whether the second folder's event comes while the first still waits or after it was read. +func TestOneMessageInTwoFoldersIsOneEntryJudgedOnce(t *testing.T) { + ctx := context.Background() + now := func() time.Time { return time.Date(2026, 10, 22, 9, 1, 0, 0, time.UTC) } + for _, tc := range []struct { + name string + readFirst bool + }{{"both waiting", false}, {"the second after the first was read", true}} { + t.Run(tc.name, func(t *testing.T) { + store := newMemStore() + judged := 0 + if err := Hear(ctx, store, heardIn("INBOX", 31, ""), now); err != nil { + t.Fatal(err) + } + if tc.readFirst { + judged += readAll(t, store) + } + if err := Hear(ctx, store, heardIn("[Gmail]/All Mail", 912, ""), now); err != nil { + t.Fatal(err) + } + judged += readAll(t, store) + entries, _ := store.Entries(ctx) + if len(entries) != 1 { + t.Fatalf("%d entries for one message, want 1: %+v", len(entries), entries) + } + if judged != 1 { + t.Fatalf("the message was judged %d times, want once", judged) + } + }) + } +} + +// 1849/AC2: a mail without a Message-ID keeps one entry per folder and UID, as before: two such mails in one +// folder are two entries, and the same one heard twice is one. +func TestAMailWithoutMessageIDIsKeyedByFolderAndUID(t *testing.T) { + ctx := context.Background() + now := func() time.Time { return time.Date(2026, 10, 22, 9, 1, 0, 0, time.UTC) } + store := newMemStore() + for _, uid := range []uint32{40, 41, 41} { + if err := Hear(ctx, store, heardIn("INBOX", uid, ""), now); err != nil { + t.Fatal(err) + } + } + if err := Hear(ctx, store, heardIn("Archive", 40, ""), now); err != nil { + t.Fatal(err) + } + readAll(t, store) + entries, _ := store.Entries(ctx) + if len(entries) != 3 { + t.Fatalf("%d entries, want 3 (INBOX 40, INBOX 41, Archive 40): %+v", len(entries), entries) + } +} + +// 1849: the key itself — the Message-ID where there is one, else folder, UIDVALIDITY and UID. +func TestMessageKey(t *testing.T) { + if MessageKey(" ", "INBOX", 7, 31) != MessageKey("", "[Gmail]/All Mail", 7, 912) { + t.Fatal("one Message-ID in two folders gives two keys") + } + if MessageKey("", "INBOX", 7, 31) == MessageKey("", "INBOX", 7, 32) { + t.Fatal("two UIDs of one folder without a Message-ID give one key") + } + if MessageKey("", "INBOX", 7, 31) == MessageKey("", "Archive", 7, 31) { + t.Fatal("one UID in two folders without a Message-ID gives one key") + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go index ad374f91..de4ed15c 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go @@ -261,10 +261,14 @@ func (s *PgStore) AddWaiting(ctx context.Context, w WaitingMail) error { 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) + // One message waits once (either unique key: the same folder and UID, or the same message key, 1849/AC1), + // and not at all once it has its entry: Gmail's All Mail copy may be heard after the INBOX copy was read. + _, err := s.pool.Exec(ctx, `insert into waiting_mail (account, folder, message_id, message_key, uidvalidity, uid, date, sender, subject, flags, size) + select $1, $2, $3, $11, $4, $5, $6, $7, $8, $9, $10 + where not exists (select 1 from mail_entry where account = $1 and message_key = $11) + on conflict do nothing`, + w.Account, w.Folder, w.MessageID, int64(w.UIDValidity), int64(w.UID), w.Date, from, w.Subject, flags, w.Size, + MessageKey(w.MessageID, w.Folder, w.UIDValidity, w.UID)) return err } @@ -355,11 +359,18 @@ func (s *PgStore) AddEntry(ctx context.Context, e Entry) (int64, error) { mark = &m } var id int64 + // One entry per account and message key (1849): a message judged again (its folder's second copy in a + // race, or the re-reading) replaces its judgement and keeps its id, headers and told. err := s.pool.QueryRow(ctx, `insert into mail_entry - (account, message_id, folder, date, sender_name, sender_address, subject, mail_group, summary, mark, why, due_date, told, judged) - values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) returning id`, + (account, message_id, folder, date, sender_name, sender_address, subject, mail_group, summary, mark, why, due_date, told, judged, + uidvalidity, uid, message_key) + values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17) + on conflict (account, message_key) do update set mail_group = excluded.mail_group, summary = excluded.summary, + mark = excluded.mark, why = excluded.why, due_date = excluded.due_date, judged = excluded.judged + returning id`, e.Account, e.MessageID, e.Folder, e.Date, e.SenderName, e.SenderAddress, e.Subject, group, e.Summary, mark, e.Why, - nullDate(e.DueDate), e.Told, e.Judged).Scan(&id) + nullDate(e.DueDate), e.Told, e.Judged, int64(e.UIDValidity), int64(e.UID), + MessageKey(e.MessageID, e.Folder, e.UIDValidity, e.UID)).Scan(&id) return id, err } @@ -392,11 +403,13 @@ func scanEntry(row pgx.Row) (Entry, error) { var group *int64 var mark *string var dueDate *time.Time + var uidvalidity, uid int64 err := row.Scan(&e.ID, &e.Account, &e.MessageID, &e.Folder, &e.Date, &e.SenderName, &e.SenderAddress, &e.Subject, - &group, &e.Summary, &mark, &e.Why, &dueDate, &e.Told, &e.Judged) + &group, &e.Summary, &mark, &e.Why, &dueDate, &e.Told, &e.Judged, &uidvalidity, &uid) if err != nil { return e, err } + e.UIDValidity, e.UID = uint32(uidvalidity), uint32(uid) if group != nil { e.Group = *group } @@ -410,7 +423,7 @@ func scanEntry(row pgx.Row) (Entry, error) { return e, nil } -const entryColumns = `id, account, message_id, folder, date, sender_name, sender_address, subject, mail_group, summary, mark, why, due_date, told, judged` +const entryColumns = `id, account, message_id, folder, date, sender_name, sender_address, subject, mail_group, summary, mark, why, due_date, told, judged, uidvalidity, uid` func (s *PgStore) Entry(ctx context.Context, id int64) (Entry, bool, error) { e, err := scanEntry(s.pool.QueryRow(ctx, `select `+entryColumns+` from mail_entry where id = $1`, id)) 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 index 2faa5b71..df51c74d 100644 --- 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 @@ -16,6 +16,7 @@ import ( "strings" "sync/atomic" "testing" + "time" "github.com/jackc/pgx/v5/pgxpool" ) @@ -177,3 +178,59 @@ func integrationSchemaShape(t *testing.T, s *PgStore) []string { } return out } + +// preparedStore is a fresh database given the consolidated schema, for the store's own integration tests. +func preparedStore(t *testing.T) *PgStore { + t.Helper() + ctx := context.Background() + s, err := OpenPgStore(ctx, freshIntegrationDatabase(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(s.Close) + if _, err := s.Prepare(ctx); err != nil { + t.Fatal(err) + } + return s +} + +// 1849/AC1 through the real store: one Gmail message heard in INBOX and in All Mail waits once, and heard +// again after its entry was kept it does not wait at all; its entry is one. +func TestPgStoreKeepsOneMessageInTwoFoldersOnce(t *testing.T) { + ctx := context.Background() + s := preparedStore(t) + date := time.Date(2026, 10, 22, 9, 0, 0, 0, time.UTC) + inbox := WaitingMail{Account: "work", Folder: "INBOX", MessageID: "", UIDValidity: 7, UID: 31, Date: date} + allMail := inbox + allMail.Folder, allMail.UID = "[Gmail]/All Mail", 912 + for _, w := range []WaitingMail{inbox, allMail} { + if err := s.AddWaiting(ctx, w); err != nil { + t.Fatal(err) + } + } + waiting, err := s.Waiting(ctx) + if err != nil || len(waiting) != 1 { + t.Fatalf("waiting = %v (%v), want the message once", waiting, err) + } + e := Entry{Account: "work", MessageID: inbox.MessageID, Folder: "INBOX", UIDValidity: 7, UID: 31, Date: date, Mark: MarkNeither, Judged: date} + if _, err := s.AddEntry(ctx, e); err != nil { + t.Fatal(err) + } + if err := s.TakeWaiting(ctx, waiting[0].Seq); err != nil { + t.Fatal(err) + } + if err := s.AddWaiting(ctx, allMail); err != nil { + t.Fatal(err) + } + if waiting, _ := s.Waiting(ctx); len(waiting) != 0 { + t.Fatalf("waiting = %v, want none: the message already has its entry", waiting) + } + e.Folder, e.UID, e.Mark = allMail.Folder, allMail.UID, MarkMatters + if _, err := s.AddEntry(ctx, e); err != nil { + t.Fatal(err) + } + entries, err := s.Entries(ctx) + if err != nil || len(entries) != 1 || entries[0].Mark != MarkMatters || entries[0].Folder != "INBOX" { + t.Fatalf("entries = %+v (%v), want one, its judgement replaced and its folder kept", entries, err) + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql b/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql index 9c592cb9..8dcf4522 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql @@ -1,4 +1,4 @@ --- The consolidated schema, through migration 0001: what a fresh database is given at once (the mesh's rule: a +-- The consolidated schema, through migration 0002: 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, @@ -10,6 +10,9 @@ create table mail_entry ( account text not null, message_id text not null default '', folder text not null, + uidvalidity bigint not null default 0, + uid bigint not null default 0, + message_key text not null default '', date timestamptz not null, sender_name text not null default '', sender_address text not null default '', @@ -20,7 +23,8 @@ create table mail_entry ( why text not null default '', due_date date, told boolean not null default false, - judged timestamptz not null default now()); + judged timestamptz not null default now(), + constraint mail_entry_one_per_message unique (account, message_key)); create index mail_entry_account_date on mail_entry (account, date desc); create index mail_entry_judged on mail_entry (judged); @@ -43,13 +47,15 @@ create index reading_lesson_sender on reading_lesson (sender_address); -- 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. +-- heard twice is kept once: by its account, folder, UIDVALIDITY and UID, and by its account and message key +-- (issue 1849: its Message-ID, else its folder, UIDVALIDITY and UID), so one Gmail message heard in INBOX and +-- in All Mail waits once. create table waiting_mail ( id bigserial primary key, account text not null, folder text not null, message_id text not null default '', + message_key text not null default '', uidvalidity bigint not null default 0, uid bigint not null default 0, date timestamptz not null, @@ -58,7 +64,8 @@ create table waiting_mail ( flags text[] not null default '{}', size bigint not null default 0, heard timestamptz not null default now(), - unique (account, folder, uidvalidity, uid)); + unique (account, folder, uidvalidity, uid), + constraint waiting_mail_one_per_message unique (account, message_key)); -- 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 diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema_model_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema_model_test.go new file mode 100644 index 00000000..09d605fe --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema_model_test.go @@ -0,0 +1,338 @@ +package main + +// The consolidated schema equals the migrations applied in order (the mesh's rule: both, and they agree; +// 1857/AC7, 1849/AC3), proven without a database: each is applied to a small model of the tables — every +// column's type, null-ness, default and inline rules, every table constraint and every index — and the two +// models compared. The model reads only the statements this module's own SQL uses (create table, create +// index, alter table add/drop/alter column and add/drop constraint); data statements (update, delete, +// insert) change no shape and are passed over; anything else fails the test, so a new kind of statement is +// taught here before it is trusted. The real postgres comparison is pgstore_integration_test.go's. + +import ( + "fmt" + "reflect" + "regexp" + "sort" + "strings" + "testing" +) + +type modelTable struct { + Columns map[string]string // name -> canonical definition + Constraints map[string]bool // canonical table constraints +} + +type schemaModel struct { + Tables map[string]*modelTable + Indexes map[string]string // name -> canonical definition +} + +func newSchemaModel() *schemaModel { + return &schemaModel{Tables: map[string]*modelTable{}, Indexes: map[string]string{}} +} + +// stripSQLComments removes -- comments outside quotes. +func stripSQLComments(sql string) string { + var b strings.Builder + inQuote := false + for i := 0; i < len(sql); i++ { + c := sql[i] + if c == '\'' { + inQuote = !inQuote + } + if !inQuote && c == '-' && i+1 < len(sql) && sql[i+1] == '-' { + for i < len(sql) && sql[i] != '\n' { + i++ + } + b.WriteByte('\n') + continue + } + b.WriteByte(c) + } + return b.String() +} + +// splitTopLevel splits s on sep where it is outside parentheses and quotes. +func splitTopLevel(s string, sep byte) []string { + var out []string + depth, inQuote, start := 0, false, 0 + for i := 0; i < len(s); i++ { + switch c := s[i]; { + case c == '\'': + inQuote = !inQuote + case inQuote: + case c == '(': + depth++ + case c == ')': + depth-- + case c == sep && depth == 0: + out = append(out, s[start:i]) + start = i + 1 + } + } + out = append(out, s[start:]) + var kept []string + for _, part := range out { + if p := strings.TrimSpace(part); p != "" { + kept = append(kept, p) + } + } + return kept +} + +var ( + spaces = regexp.MustCompile(`\s+`) + aroundPunct = regexp.MustCompile(`\s*([(),])\s*`) + createTable = regexp.MustCompile(`^create table (?:if not exists )?(\w+)\s*\((.*)\)$`) + createIndex = regexp.MustCompile(`^create (unique )?index (?:if not exists )?(\w+) on (\w+)\s*(.*)$`) + alterTable = regexp.MustCompile(`^alter table (\w+) (.*)$`) + addColumn = regexp.MustCompile(`^add column (?:if not exists )?(.*)$`) + alterColumn = regexp.MustCompile(`^alter column (\w+) (set not null|drop not null|drop default|set default (.*))$`) + addConstr = regexp.MustCompile(`^add (constraint \w+ .*)$`) + dropConstr = regexp.MustCompile(`^drop constraint (?:if exists )?(\w+)$`) + dropColumn = regexp.MustCompile(`^drop column (?:if exists )?(\w+)$`) + constraintNm = regexp.MustCompile(`^constraint (\w+) `) +) + +func canonicalSQL(s string) string { + s = strings.ToLower(strings.TrimSpace(spaces.ReplaceAllString(s, " "))) + return aroundPunct.ReplaceAllString(s, "$1") +} + +// column is one column definition, canonical: its type, then "not null", then "default ", then whatever +// else it carries (primary key, unique, references, check), in that order whatever order it was written in. +type column struct { + Type, Default, Rest string + NotNull bool +} + +func (c column) String() string { + return fmt.Sprintf("%s|notnull=%v|default=%s|%s", c.Type, c.NotNull, c.Default, c.Rest) +} + +var columnKeywords = map[string]bool{"not": true, "null": true, "default": true, "primary": true, "unique": true, + "references": true, "check": true, "constraint": true, "generated": true} + +func parseColumn(def string) (string, column) { + words := splitWords(canonicalSQL(def)) + name := words[0] + var c column + i := 1 + var typ []string + for ; i < len(words) && !columnKeywords[words[i]]; i++ { + typ = append(typ, words[i]) + } + c.Type = strings.Join(typ, " ") + var rest []string + for i < len(words) { + switch { + case words[i] == "not" && i+1 < len(words) && words[i+1] == "null": + c.NotNull = true + i += 2 + case words[i] == "null": + i++ + case words[i] == "default": + c.Default = words[i+1] + i += 2 + default: + rest = append(rest, words[i]) + i++ + } + } + c.Rest = strings.Join(rest, " ") + return name, c +} + +// splitWords splits a canonical definition on spaces outside parentheses and quotes, a parenthesised group +// staying with the word before it ("references mail_group(id)" -> references, mail_group(id)). +func splitWords(s string) []string { return splitTopLevel(s, ' ') } + +func (m *schemaModel) table(t *testing.T, name string) *modelTable { + t.Helper() + tb := m.Tables[name] + if tb == nil { + t.Fatalf("the table %s is changed before it is made", name) + } + return tb +} + +// apply applies one file's statements to the model. +func (m *schemaModel) apply(t *testing.T, file, sql string) { + t.Helper() + for _, stmt := range splitTopLevel(stripSQLComments(sql), ';') { + s := strings.ToLower(strings.TrimSpace(spaces.ReplaceAllString(stmt, " "))) + switch { + case createTable.MatchString(s): + g := createTable.FindStringSubmatch(s) + tb := &modelTable{Columns: map[string]string{}, Constraints: map[string]bool{}} + for _, item := range splitTopLevel(g[2], ',') { + first := strings.Fields(item)[0] + if first == "constraint" || first == "unique" || first == "primary" || first == "check" || first == "foreign" { + tb.Constraints[canonicalSQL(item)] = true + continue + } + name, c := parseColumn(item) + tb.Columns[name] = c.String() + } + m.Tables[g[1]] = tb + case createIndex.MatchString(s): + g := createIndex.FindStringSubmatch(s) + m.Indexes[g[2]] = canonicalSQL(g[1] + "on " + g[3] + " " + g[4]) + case strings.HasPrefix(s, "drop index "): + delete(m.Indexes, strings.TrimPrefix(strings.TrimPrefix(s, "drop index "), "if exists ")) + case alterTable.MatchString(s): + g := alterTable.FindStringSubmatch(s) + tb := m.table(t, g[1]) + for _, action := range splitTopLevel(g[2], ',') { + m.alter(t, file, tb, action) + } + case strings.HasPrefix(s, "update ") || strings.HasPrefix(s, "delete ") || strings.HasPrefix(s, "insert "): + // data, not shape + default: + t.Fatalf("%s: a statement the schema model does not read yet: %q", file, s) + } + } +} + +func (m *schemaModel) alter(t *testing.T, file string, tb *modelTable, action string) { + t.Helper() + switch { + case addColumn.MatchString(action): + name, c := parseColumn(addColumn.FindStringSubmatch(action)[1]) + tb.Columns[name] = c.String() + case dropColumn.MatchString(action): + delete(tb.Columns, dropColumn.FindStringSubmatch(action)[1]) + case alterColumn.MatchString(action): + g := alterColumn.FindStringSubmatch(action) + name, change := g[1], g[2] + old, ok := tb.Columns[name] + if !ok { + t.Fatalf("%s: alters the column %s, which does not exist", file, name) + } + var c column + parts := strings.SplitN(old, "|", 4) + c.Type, c.NotNull, c.Default, c.Rest = parts[0], parts[1] == "notnull=true", strings.TrimPrefix(parts[2], "default="), parts[3] + switch { + case change == "set not null": + c.NotNull = true + case change == "drop not null": + c.NotNull = false + case change == "drop default": + c.Default = "" + default: + c.Default = canonicalSQL(g[3]) + } + tb.Columns[name] = c.String() + case addConstr.MatchString(action): + tb.Constraints[canonicalSQL(addConstr.FindStringSubmatch(action)[1])] = true + case dropConstr.MatchString(action): + name := dropConstr.FindStringSubmatch(action)[1] + for k := range tb.Constraints { + if g := constraintNm.FindStringSubmatch(k); g != nil && g[1] == name { + delete(tb.Constraints, k) + } + } + default: + t.Fatalf("%s: an alter table the schema model does not read yet: %q", file, action) + } +} + +func describeModel(m *schemaModel) []string { + var out []string + for name, tb := range m.Tables { + for col, def := range tb.Columns { + out = append(out, name+"."+col+" "+def) + } + for c := range tb.Constraints { + out = append(out, name+" "+c) + } + } + for name, def := range m.Indexes { + out = append(out, "index "+name+" "+def) + } + sort.Strings(out) + return out +} + +func modelOfMigrations(t *testing.T) *schemaModel { + t.Helper() + ms, err := Migrations() + if err != nil { + t.Fatal(err) + } + m := newSchemaModel() + for _, mig := range ms { + m.apply(t, fmt.Sprintf("%04d-%s.sql", mig.Number, mig.Name), mig.SQL) + } + return m +} + +func modelOfConsolidated(t *testing.T) *schemaModel { + t.Helper() + m := newSchemaModel() + m.apply(t, "schema.sql", consolidated) + return m +} + +// 1857/AC7 (and the rule for every migration): schema.sql equals the migrations applied in order, and says +// how far it is consolidated. +func TestTheConsolidatedSchemaEqualsTheMigrationsInOrder(t *testing.T) { + a, b := describeModel(modelOfConsolidated(t)), describeModel(modelOfMigrations(t)) + if !reflect.DeepEqual(a, b) { + onlyA, onlyB := difference(a, b), difference(b, a) + t.Fatalf("schema.sql and the migrations differ:\nonly in schema.sql: %v\nonly in the migrations: %v", onlyA, onlyB) + } + ms, _ := Migrations() + if want := fmt.Sprintf("through migration %04d", len(ms)); !strings.Contains(consolidated, want) { + t.Fatalf("schema.sql does not say %q", want) + } +} + +func difference(a, b []string) []string { + in := map[string]bool{} + for _, s := range b { + in[s] = true + } + var out []string + for _, s := range a { + if !in[s] { + out = append(out, s) + } + } + return out +} + +// The model itself reads a column the same whether it was made whole or added and altered. +func TestTheSchemaModelReadsAnAlteredColumnAsAWholeOne(t *testing.T) { + whole, stepped := newSchemaModel(), newSchemaModel() + whole.apply(t, "whole", "create table a (id bigserial primary key, k text not null default '', constraint a_k unique (k));") + stepped.apply(t, "stepped", "create table a (id bigserial primary key);\nalter table a add column k text;\n"+ + "update a set k = '';\nalter table a alter column k set not null;\nalter table a alter column k set default '';\n"+ + "alter table a add constraint a_k unique (k);") + if a, b := describeModel(whole), describeModel(stepped); !reflect.DeepEqual(a, b) { + t.Fatalf("whole %v\nstepped %v", a, b) + } + other := newSchemaModel() + other.apply(t, "other", "create table a (id bigserial primary key, k text not null default '');") + if reflect.DeepEqual(describeModel(whole), describeModel(other)) { + t.Fatal("the model does not see a missing constraint") + } +} + +// 1849/AC3: a unique constraint per account and message key holds on mail_entry and waiting_mail, in +// schema.sql and in the migrations. +func TestOneEntryAndOneWaitingMailPerAccountAndMessageKey(t *testing.T) { + for which, m := range map[string]*schemaModel{"schema.sql": modelOfConsolidated(t), "the migrations": modelOfMigrations(t)} { + for _, table := range []string{"mail_entry", "waiting_mail"} { + found := false + for c := range m.Tables[table].Constraints { + if strings.HasSuffix(c, "unique(account,message_key)") { + found = true + } + } + if !found { + t.Errorf("%s: %s has no unique constraint on (account, message_key): %v", which, table, m.Tables[table].Constraints) + } + } + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go index a93c6575..7afd9565 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go @@ -6,6 +6,8 @@ package main import ( "context" + "fmt" + "strings" "time" ) @@ -31,6 +33,8 @@ type Entry struct { Account string MessageID string Folder string + UIDValidity uint32 // the folder's UIDVALIDITY and the mail's UID: its key when it has no Message-ID (1849) + UID uint32 Date time.Time SenderName string SenderAddress string @@ -94,14 +98,11 @@ type WaitingMail struct { 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). +// Key is the waiting mail's own identity: its account and MessageKey (issue 1849) — its message id wherever it +// has one (stable across a UIDVALIDITY reset and across Gmail's folders), else its folder, UIDVALIDITY and UID +// (novox/hq ADR 0459 §3). func (w WaitingMail) Key() string { - if w.MessageID != "" { - return w.Account + "\x1f" + w.MessageID - } - return w.Account + "\x1f" + w.Folder + return w.Account + "\x1f" + MessageKey(w.MessageID, w.Folder, w.UIDValidity, w.UID) } // HealthOutcome names why a batch's judgement did not succeed (novox/hq ADR 0459 §12); the empty outcome means @@ -130,7 +131,8 @@ type ReaderHealth struct { 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. + // matched by WaitingMail.Key: the same message in a second folder too, 1849/AC1) is recorded once, and a + // mail that already has its entry is not recorded again; 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) @@ -142,7 +144,8 @@ type Store interface { // SetReaderHealth replaces the kept health state. SetReaderHealth(ctx context.Context, h ReaderHealth) error - // AddEntry records a new mail entry (1520/AC3) and answers its own id. + // AddEntry records a mail entry (1520/AC3) and answers its own id: one per account and MessageKey (1849) — + // for a message that already has one, its judgement is replaced and its id kept. AddEntry(ctx context.Context, e Entry) (int64, error) // UpdateEntry replaces an entry's fields by its id (1428/AC1, correct). ErrNotFound when unknown. UpdateEntry(ctx context.Context, e Entry) error @@ -173,3 +176,14 @@ var ErrNotFound = notFoundError{} type notFoundError struct{} func (notFoundError) Error() string { return "not found" } + +// MessageKey is a mail's identity within its account (issue 1849): its Message-ID where it has one — Gmail +// files one message in several folders (INBOX, All Mail), each with its own UID, and the Message-ID is the +// same in each — else its folder, UIDVALIDITY and UID. waiting_mail and mail_entry are each unique on the +// account and this key (schema.sql), so one message is one waiting mail and one entry, judged once. +func MessageKey(messageID, folder string, uidvalidity, uid uint32) string { + if id := strings.TrimSpace(messageID); id != "" { + return "mid:" + id + } + return fmt.Sprintf("uid:%s/%d/%d", folder, uidvalidity, uid) +}