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 } // ---- entries, groups and reading lessons (1427, 1520, 1521, 1428) ---------------------------------------------- func nullDate(t time.Time) *time.Time { if t.IsZero() { return nil } return &t } func (s *PgStore) AddEntry(ctx context.Context, e Entry) (int64, error) { var group *int64 if e.Group != 0 { group = &e.Group } var mark *string if e.Mark != "" { m := string(e.Mark) mark = &m } var id int64 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`, 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) return id, err } func (s *PgStore) UpdateEntry(ctx context.Context, e Entry) error { var group *int64 if e.Group != 0 { group = &e.Group } var mark *string if e.Mark != "" { m := string(e.Mark) mark = &m } tag, err := s.pool.Exec(ctx, `update mail_entry set account = $2, message_id = $3, folder = $4, date = $5, sender_name = $6, sender_address = $7, subject = $8, mail_group = $9, summary = $10, mark = $11, why = $12, due_date = $13, told = $14, judged = $15 where id = $1`, e.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) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } func scanEntry(row pgx.Row) (Entry, error) { var e Entry var group *int64 var mark *string var dueDate *time.Time 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) if err != nil { return e, err } if group != nil { e.Group = *group } if mark != nil { e.Mark = Mark(*mark) } if dueDate != nil { e.DueDate = *dueDate } e.Date, e.Judged = e.Date.UTC(), e.Judged.UTC() 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` 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)) if errors.Is(err, pgx.ErrNoRows) { return Entry{}, false, nil } return e, err == nil, err } func (s *PgStore) Entries(ctx context.Context) ([]Entry, error) { rows, err := s.pool.Query(ctx, `select `+entryColumns+` from mail_entry`) if err != nil { return nil, err } defer rows.Close() var out []Entry for rows.Next() { e, err := scanEntry(rows) if err != nil { return nil, err } out = append(out, e) } return out, rows.Err() } func (s *PgStore) FindOrCreateGroup(ctx context.Context, label string) (Group, error) { var g Group err := s.pool.QueryRow(ctx, `insert into mail_group (label) values ($1) on conflict (label) do nothing returning id, label, created`, label).Scan(&g.ID, &g.Label, &g.Created) if errors.Is(err, pgx.ErrNoRows) { err = s.pool.QueryRow(ctx, `select id, label, created from mail_group where label = $1`, label).Scan(&g.ID, &g.Label, &g.Created) } if err != nil { return Group{}, err } g.Created = g.Created.UTC() return g, nil } func (s *PgStore) Groups(ctx context.Context) ([]Group, error) { rows, err := s.pool.Query(ctx, `select g.id, g.label, g.created, count(e.id) from mail_group g left join mail_entry e on e.mail_group = g.id group by g.id, g.label, g.created`) if err != nil { return nil, err } defer rows.Close() var out []Group for rows.Next() { var g Group if err := rows.Scan(&g.ID, &g.Label, &g.Created, &g.Count); err != nil { return nil, err } g.Created = g.Created.UTC() out = append(out, g) } return out, rows.Err() } func (s *PgStore) AddLesson(ctx context.Context, l Lesson) (int64, error) { var group *int64 if l.Group != 0 { group = &l.Group } var id int64 err := s.pool.QueryRow(ctx, `insert into reading_lesson (account, folder, sender_name, sender_address, subject, mail_group, change, made, undone) values ($1, $2, $3, $4, $5, $6, $7, $8, $9) returning id`, l.Account, l.Folder, l.SenderName, l.SenderAddress, l.Subject, group, l.Change, l.Made, l.Undone).Scan(&id) return id, err } func (s *PgStore) Lessons(ctx context.Context) ([]Lesson, error) { rows, err := s.pool.Query(ctx, `select id, account, folder, sender_name, sender_address, subject, mail_group, change, made, undone from reading_lesson`) if err != nil { return nil, err } defer rows.Close() var out []Lesson for rows.Next() { var l Lesson var group *int64 if err := rows.Scan(&l.ID, &l.Account, &l.Folder, &l.SenderName, &l.SenderAddress, &l.Subject, &group, &l.Change, &l.Made, &l.Undone); err != nil { return nil, err } if group != nil { l.Group = *group } l.Made = l.Made.UTC() out = append(out, l) } return out, rows.Err() } func (s *PgStore) UndoLesson(ctx context.Context, id int64) error { tag, err := s.pool.Exec(ctx, `update reading_lesson set undone = true where id = $1`, id) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } return nil }