519 lines
16 KiB
Go
519 lines
16 KiB
Go
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
|
|
}
|