issue 1849: keep one mail entry per message, by its Message-ID, else its folder and UID
Gmail files one message in INBOX and All Mail, each with its own UID; the reader kept an entry per folder. waiting_mail and mail_entry are now unique per account and message key (migration 0002 and schema.sql), a mail that already has its entry no longer waits again, and the test double keeps PgStore's own keys (issue 1548). A schema model test proves schema.sql equals the migrations in order.
This commit is contained in:
jochen
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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" },
|
||||
|
||||
@@ -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()
|
||||
|
||||
+22
@@ -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);
|
||||
@@ -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, "<abc@example.com>"), now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tc.readFirst {
|
||||
judged += readAll(t, store)
|
||||
}
|
||||
if err := Hear(ctx, store, heardIn("[Gmail]/All Mail", 912, "<abc@example.com>"), 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(" <abc@example.com> ", "INBOX", 7, 31) != MessageKey("<abc@example.com>", "[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")
|
||||
}
|
||||
}
|
||||
@@ -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))
|
||||
|
||||
@@ -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: "<abc@example.com>", 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 <x>", 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user