mesh-catalog/modules/mesh-mail/cmd/mesh-mail/pgstore.go

package main

// The index on the mesh's postgres (novox/hq ADR 0359 §3): 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 prepare`, ADR 0135) 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: the keeper, the watch and the tools refuse a schema that is behind, and say so.

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.
func Migrations() ([]Migration, error) {
	entries, err := fs.ReadDir(migrationFiles, "migrations")
	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(migrationFiles, path.Join("migrations", 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.
const lockKey int64 = 4_250_359_001

// PgStore is the Store 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.
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('message') is not null`).Scan(&exists); err != nil {
			return nil, err
		}
		if exists {
			return nil, errors.New("the database holds the index'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 prepare) has not run", at, len(migrations))
	}
	return nil
}

func (s *PgStore) Folder(ctx context.Context, account, folder string) (Folder, bool, error) {
	f := Folder{Account: account, Name: folder}
	var validity int64
	err := s.pool.QueryRow(ctx, `select uidvalidity, messages, synced from folder where account = $1 and name = $2`, account, folder).
		Scan(&validity, &f.Messages, &f.Synced)
	if errors.Is(err, pgx.ErrNoRows) {
		return f, false, nil
	}
	f.UIDValidity = uint32(validity)
	return f, err == nil, err
}

func (s *PgStore) SetFolder(ctx context.Context, f Folder) error {
	_, err := s.pool.Exec(ctx, `insert into folder (account, name, uidvalidity, messages, synced) values ($1, $2, $3, $4, $5)
		on conflict (account, name) do update set uidvalidity = excluded.uidvalidity, messages = excluded.messages, synced = excluded.synced`,
		f.Account, f.Name, int64(f.UIDValidity), f.Messages, f.Synced)
	return err
}

func (s *PgStore) DropFolder(ctx context.Context, account, folder string) error {
	return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
		if _, err := tx.Exec(ctx, `delete from message where account = $1 and folder = $2`, account, folder); err != nil {
			return err
		}
		_, err := tx.Exec(ctx, `delete from folder where account = $1 and name = $2`, account, folder)
		return err
	})
}

func (s *PgStore) Flags(ctx context.Context, account, folder string) (map[uint32][]string, error) {
	rows, err := s.pool.Query(ctx, `select uid, flags from message where account = $1 and folder = $2`, account, folder)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[uint32][]string{}
	for rows.Next() {
		var uid int64
		var flags []string
		if err := rows.Scan(&uid, &flags); err != nil {
			return nil, err
		}
		out[uint32(uid)] = flags
	}
	return out, rows.Err()
}

func keepEvent(ctx context.Context, tx pgx.Tx, e *Event) error {
	if e == nil {
		return nil
	}
	body, err := json.Marshal(e.Body)
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `insert into outbox (key, body) values ($1, $2)`, e.Key, body)
	return err
}

func (s *PgStore) Add(ctx context.Context, m Message, e *Event) error {
	from, _ := json.Marshal(m.From)
	to, _ := json.Marshal(m.To)
	cc, _ := json.Marshal(orNone(m.Cc))
	atts, _ := json.Marshal(m.Attachments)
	flags := m.Flags
	if flags == nil {
		flags = []string{}
	}
	people := strings.Join([]string{peopleText(m.From), peopleText(m.To), peopleText(m.Cc)}, " ")
	// An address is one word to the search, so its parts are kept beside it: "bob" finds bob@example.org.
	people += " " + strings.NewReplacer("@", " ", ".", " ", ",", " ").Replace(people)
	return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
		if _, err := tx.Exec(ctx, `insert into message (account, folder, uid, uidvalidity, message_id, in_reply_to, date, sender,
			recipients, copied, people, subject, size, flags, attachments, body)
			values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
			on conflict (account, folder, uid) do update set uidvalidity = excluded.uidvalidity, message_id = excluded.message_id,
			in_reply_to = excluded.in_reply_to, date = excluded.date, sender = excluded.sender, recipients = excluded.recipients,
			copied = excluded.copied, people = excluded.people, subject = excluded.subject, size = excluded.size,
			flags = excluded.flags, attachments = excluded.attachments, body = excluded.body`,
			m.Account, m.Folder, int64(m.UID), int64(m.UIDValidity), m.MessageID, m.InReplyTo, m.Date, from, to, cc, people,
			strings.ToValidUTF8(strings.ReplaceAll(m.Subject, "\x00", ""), ""), m.Size, flags, atts,
			strings.ReplaceAll(m.Body, "\x00", "")); err != nil {
			return err
		}
		return keepEvent(ctx, tx, e)
	})
}

func orNone(ps []Person) []Person {
	if ps == nil {
		return []Person{}
	}
	return ps
}

func (s *PgStore) SetFlags(ctx context.Context, account, folder string, uid uint32, flags []string, e *Event) error {
	return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
		if _, err := tx.Exec(ctx, `update message set flags = $4 where account = $1 and folder = $2 and uid = $3`,
			account, folder, int64(uid), flags); err != nil {
			return err
		}
		return keepEvent(ctx, tx, e)
	})
}

func (s *PgStore) Remove(ctx context.Context, account, folder string, uid uint32, e *Event) error {
	return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
		if _, err := tx.Exec(ctx, `delete from message where account = $1 and folder = $2 and uid = $3`, account, folder, int64(uid)); err != nil {
			return err
		}
		return keepEvent(ctx, tx, e)
	})
}

const messageColumns = `account, folder, uid, uidvalidity, message_id, in_reply_to, date, sender, recipients, copied, subject,
	size, flags, attachments`

func scanMessage(row pgx.Row, withBody bool) (Message, error) {
	var m Message
	var uid, validity int64
	var from, to, cc, atts []byte
	dest := []any{&m.Account, &m.Folder, &uid, &validity, &m.MessageID, &m.InReplyTo, &m.Date, &from, &to, &cc, &m.Subject,
		&m.Size, &m.Flags, &atts}
	if withBody {
		dest = append(dest, &m.Body)
	}
	if err := row.Scan(dest...); err != nil {
		return m, err
	}
	m.UID, m.UIDValidity, m.Date = uint32(uid), uint32(validity), m.Date.UTC()
	for _, f := range []struct {
		raw  []byte
		into any
	}{{from, &m.From}, {to, &m.To}, {cc, &m.Cc}, {atts, &m.Attachments}} {
		if err := json.Unmarshal(f.raw, f.into); err != nil {
			return m, err
		}
	}
	return m, nil
}

func (s *PgStore) Get(ctx context.Context, account, folder string, uid uint32) (Message, bool, error) {
	m, err := scanMessage(s.pool.QueryRow(ctx, `select `+messageColumns+`, body from message where account = $1 and folder = $2 and uid = $3`,
		account, folder, int64(uid)), true)
	if errors.Is(err, pgx.ErrNoRows) {
		return Message{}, false, nil
	}
	return m, err == nil, err
}

// addressFilterSQL is 1228/AC2, 1228/AC4, 1228/AC6(cc): an EXISTS over the jsonb array column (recipients or
// copied), matched against one already-lowercased parameter either as a whole address or, with no "@", as a
// domain (every address under it): right(address, length(domain)+1) = '@'||domain, an exact substring compare,
// not LIKE, so a filter holding "%" or "_" is never a wildcard (issue 1228, review round 1 finding 4: a filter
// from:["%"] matched every mail in Postgres and none in memStore, since memStore's strings.HasSuffix never
// treats "%" specially). column names a fixed set the Go code chooses, never caller input.
func addressFilterSQL(column, param string) string {
	return fmt.Sprintf("exists (select 1 from jsonb_array_elements(%s) el where lower(el->>'address') = %s or "+
		"right(lower(el->>'address'), length(%s) + 1) = '@' || %s)", column, param, param, param)
}

// addressFilterSQLAny is 1228/AC3; 1230/AC11 ("several domains"): an EXISTS over column (sender), matched
// against any of filters, OR'd (each already-lowercased, each either a whole address or, with no "@", a domain),
// the same exact substring compare as addressFilterSQL, never LIKE. The whole list reaches SQL as one
// parameter, a text[]; never concatenated.
func addressFilterSQLAny(column string, filters []string, param func(any) string) string {
	lowered := make([]string, len(filters))
	for i, f := range filters {
		lowered[i] = strings.ToLower(f)
	}
	p := param(lowered)
	return fmt.Sprintf("exists (select 1 from jsonb_array_elements(%s) el, unnest(%s::text[]) f where "+
		"lower(el->>'address') = f or right(lower(el->>'address'), length(f) + 1) = '@' || f)", column, p)
}

// wordFilterSQL is 1229/AC13 (one column: SubjectWords or BodyWords); AC5, AC6(words), AC12, AC15 (review round
// 1 finding 3: columns holds both subject and body for Words, so a word counts in either — "the word in subject
// or body"). One condition per word of the list, each a whole-word, case-insensitive match (Postgres's \y, its
// own word boundary) against any one of columns, so a hyphenated word such as "4711-882" is never matched by
// only its digits (AC12); the per-word conditions join with "and" ("all", the default) or "or" ("any"). Empty
// answers "" (no filter). The word itself always reaches SQL as param's parameter, never concatenated; only its
// \y anchors and the column names (a fixed set the Go code chooses) are fixed SQL text.
func wordFilterSQL(columns []string, words []string, any bool, param func(any) string) string {
	if len(words) == 0 {
		return ""
	}
	parts := make([]string, 0, len(words))
	for _, w := range words {
		p := param(quoteRegex(w))
		cols := make([]string, len(columns))
		for i, col := range columns {
			cols[i] = fmt.Sprintf(`%s ~* ('\y' || %s || '\y')`, col, p)
		}
		parts = append(parts, "("+strings.Join(cols, " or ")+")")
	}
	op := " and "
	if any {
		op = " or "
	}
	return "(" + strings.Join(parts, op) + ")"
}

// quoteRegex escapes a word for Postgres's regex operators (ARE, a superset of POSIX ERE): the word is always
// matched literally, never as a pattern of its own, mirroring Go's regexp.QuoteMeta used by memStore's own match.
var quoteRegexEscaper = strings.NewReplacer(
	`\`, `\\`, ".", `\.`, "+", `\+`, "*", `\*`, "?", `\?`, "(", `\(`, ")", `\)`,
	"[", `\[`, "]", `\]`, "{", `\{`, "}", `\}`, "^", `\^`, "$", `\$`, "|", `\|`)

func quoteRegex(word string) string { return quoteRegexEscaper.Replace(word) }

// globToLike turns a shell glob ("*" and "?") into a SQL LIKE pattern, escaping LIKE's own metacharacters first
// (issue 1229, AC5, AC9: an attachment name pattern such as "*.pdf").
func globToLike(pattern string) string {
	var b strings.Builder
	for _, r := range pattern {
		switch r {
		case '*':
			b.WriteByte('%')
		case '?':
			b.WriteByte('_')
		case '%', '_', '\\':
			b.WriteByte('\\')
			b.WriteRune(r)
		default:
			b.WriteRune(r)
		}
	}
	return b.String()
}

// lowerStrings is 1231/AC14 (review round 1 finding 1): sentFolderNames, lowercased once, to bind as a single
// text[] parameter against lower(m2.folder) = any(...).
func lowerStrings(ss []string) []string {
	out := make([]string, len(ss))
	for i, s := range ss {
		out[i] = strings.ToLower(s)
	}
	return out
}

// searchWhere is every filter of Query (everything but its order and its limit), shared by Search and
// GroupBySender so the two can never drift apart; every value from the query reaches SQL as one of args'
// parameters, never concatenated (only fixed column names and SQL text are built as strings).
func searchWhere(q Query) (whereSQL string, args []any) {
	param := func(v any) string {
		args = append(args, v)
		return fmt.Sprintf("$%d", len(args))
	}
	where := []string{"account = any(" + param(q.Accounts) + ")"}
	if q.Folder != "" {
		where = append(where, "folder = "+param(q.Folder))
	}
	if q.Text != "" {
		where = append(where, "search @@ plainto_tsquery('simple'::regconfig, "+param(q.Text)+")")
	}
	if len(q.From) > 0 {
		where = append(where, addressFilterSQLAny("sender", q.From, param))
	}
	if q.To != "" {
		where = append(where, addressFilterSQL("recipients", param(strings.ToLower(q.To))))
	}
	if q.Cc != "" {
		where = append(where, addressFilterSQL("copied", param(strings.ToLower(q.Cc))))
	}
	if !q.OnOrAfter.IsZero() {
		where = append(where, "date >= "+param(q.OnOrAfter))
	}
	if !q.Before.IsZero() {
		where = append(where, "date < "+param(q.Before))
	}
	if c := wordFilterSQL([]string{"subject"}, q.SubjectWords, q.SubjectWordsAny, param); c != "" {
		where = append(where, c)
	}
	if c := wordFilterSQL([]string{"body"}, q.BodyWords, q.BodyWordsAny, param); c != "" {
		where = append(where, c)
	}
	// 1229/AC5, AC6(words), AC12, AC15 (review round 1 finding 3): a word found in subject or body.
	if c := wordFilterSQL([]string{"subject", "body"}, q.Words, q.WordsAny, param); c != "" {
		where = append(where, c)
	}
	if q.HasAttachment {
		where = append(where, "jsonb_array_length(attachments) > 0")
	}
	if q.AttachmentType != "" || q.AttachmentNamePattern != "" {
		var conds []string
		if q.AttachmentType != "" {
			conds = append(conds, "lower(el->>'type') = "+param(strings.ToLower(q.AttachmentType)))
		}
		if q.AttachmentNamePattern != "" {
			conds = append(conds, "lower(el->>'name') like "+param(globToLike(strings.ToLower(q.AttachmentNamePattern))))
		}
		where = append(where, "exists (select 1 from jsonb_array_elements(attachments) el where "+strings.Join(conds, " or ")+")")
	}
	if q.Unread != nil {
		where = append(where, fmt.Sprintf("(%s = any(flags)) = %s", param(`\Seen`), param(!*q.Unread)))
	}
	if q.InReplyTo != "" {
		where = append(where, "in_reply_to = "+param(q.InReplyTo))
	}
	if q.FromSomeoneIWroteTo {
		// 1231/AC14 (review round 1 finding 1): some sender address of this message is a recipient (to or cc) of
		// a mail, in the SAME account (m2.account = message.account), from a folder named one of sentFolderNames
		// ("Sent" for mesh and mediahuis/mailu, "[Gmail]/Sent Mail" for gmail; a future account naming it a third
		// way is the tool layer's "say what you chose", not this query). The list reaches SQL as one parameter.
		where = append(where, fmt.Sprintf(`exists (select 1 from jsonb_array_elements(sender) s where exists (
			select 1 from message m2, jsonb_array_elements(m2.recipients || m2.copied) r
			where m2.account = message.account and lower(m2.folder) = any(%s) and lower(r->>'address') = lower(s->>'address')))`,
			param(lowerStrings(sentFolderNames))))
	}
	return strings.Join(where, " and "), args
}

// orderColumns is 1230's fixed Go map from an order-by name to its (safe, never caller-built) SQL expression;
// unknown or empty answers "date" — the tool refuses an unknown order-by before the store is ever asked (AC16).
var orderColumns = map[string]string{
	"date":    "date",
	"sender":  "lower(coalesce(sender->0->>'address', ''))",
	"subject": "subject",
	"size":    "size",
}

func orderExpr(orderBy string) string {
	if e, ok := orderColumns[orderBy]; ok {
		return e
	}
	return orderColumns["date"]
}

// dedupedMessagesSQL is 1248/AC1, AC2, AC4-AC7: a FROM-clause subquery, aliased "message" so the caller needs no
// further change, that answers each (account, message-id) once among the rows whereSQL matches: grouped by
// account and message_id, except an empty message_id (AC5) is never grouped with anything else — each such row
// gets its own one-row group, keyed by its own (account, folder, uid) instead, which is already unique. Within a
// group the copy outside archiveFolderNames wins (AC1); only when every copy is archived does the archive copy
// answer (AC2). When a message-id sits in two non-archive folders too (e.g. INBOX and a label folder such as
// "INBOX/Telenet"; review round 1 finding 1), folder ascending then uid ascending breaks the tie — the same
// fixed rule memStore's dedupeBetter (store.go) uses, so the two stores never pick a different copy. Callers
// skip this (keep the plain "message where ...") when a folder is named (AC3): every mail of that folder
// answers, never merged with anything else.
func dedupedMessagesSQL(whereSQL string, param func(any) string) string {
	archiveParam := param(lowerStrings(archiveFolderNames))
	return `(select distinct on (account, dedup_key) ` + messageColumns + ` from (
		select *,
			case when message_id = '' then 'u:' || folder || ':' || uid::text else message_id end as dedup_key,
			(lower(folder) = any(` + archiveParam + `)) as dedup_archived
		from message where ` + whereSQL + `
	) m order by account, dedup_key, dedup_archived asc, folder asc, uid asc) message`
}

func (s *PgStore) Search(ctx context.Context, q Query) ([]Message, error) {
	limit := q.Limit
	if limit <= 0 {
		limit = 20
	}
	whereSQL, args := searchWhere(q)
	param := func(v any) string {
		args = append(args, v)
		return fmt.Sprintf("$%d", len(args))
	}
	// 1230 (AC7, AC11, AC17's schema): order by the chosen expression, ascending unless desc; an empty OrderBy
	// keeps the newest-first default (AC1) regardless of OrderDesc. Ties always break ascending on
	// (account, folder, uid), so the order is stable.
	dir := "asc"
	if q.OrderBy == "" || q.OrderDesc {
		dir = "desc"
	}
	// 1248/AC1-AC7: with no folder named, order/limit/group-by apply to the de-duplicated set (dedupedMessagesSQL);
	// a named folder (AC3) keeps the plain scan.
	fromSQL := "message where " + whereSQL
	if q.Folder == "" {
		fromSQL = dedupedMessagesSQL(whereSQL, param)
	}
	sqlText := "select " + messageColumns + " from " + fromSQL +
		" order by " + orderExpr(q.OrderBy) + " " + dir + ", account, folder, uid limit " + param(limit)
	rows, err := s.pool.Query(ctx, sqlText, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Message
	for rows.Next() {
		m, err := scanMessage(rows, false)
		if err != nil {
			return nil, err
		}
		out = append(out, m)
	}
	return out, rows.Err()
}

// GroupBySender is 1230/AC10: one line per sender among the messages searchWhere's filters match, by count
// descending (address ascending to break ties), Limit bounding the number of lines.
func (s *PgStore) GroupBySender(ctx context.Context, q Query) ([]SenderGroup, error) {
	limit := q.Limit
	if limit <= 0 {
		limit = 20
	}
	whereSQL, args := searchWhere(q)
	param := func(v any) string {
		args = append(args, v)
		return fmt.Sprintf("$%d", len(args))
	}
	// 1248/AC4: with no folder named, group by sender over the de-duplicated set (dedupedMessagesSQL), so a mail
	// held in two folders counts once; a named folder keeps the plain scan, unchanged.
	fromSQL, groupWhere := "message", whereSQL
	if q.Folder == "" {
		fromSQL, groupWhere = dedupedMessagesSQL(whereSQL, param), "true"
	}
	// Review round 1 finding 6: grouped by lower(address) (not case-sensitively, as a mixed-case repeat of the
	// same address would otherwise split into two lines); name and address are the newest mail's own
	// (array_agg ordered by date, its first element), never max()'s lexicographically largest, which is neither.
	// The outer select re-sorts by lower(address): Postgres resolves a SELECT alias used bare in order by, but
	// not wrapped in a function, so the grouping is wrapped in its own subquery first.
	sqlText := `select address, name, count, newest from (
		select (array_agg(el->>'address' order by date desc))[1] as address,
			(array_agg(el->>'name' order by date desc))[1] as name, count(*) as count, max(date) as newest
		from ` + fromSQL + `, jsonb_array_elements(sender) el where ` + groupWhere + ` group by lower(el->>'address')
		) grouped order by count desc, lower(address) asc limit ` + param(limit)
	rows, err := s.pool.Query(ctx, sqlText, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []SenderGroup
	for rows.Next() {
		var g SenderGroup
		var name *string
		if err := rows.Scan(&g.Address, &name, &g.Count, &g.Newest); err != nil {
			return nil, err
		}
		if name != nil {
			g.Name = *name
		}
		g.Newest = g.Newest.UTC()
		out = append(out, g)
	}
	return out, rows.Err()
}

func (s *PgStore) Folders(ctx context.Context, account string) ([]Folder, error) {
	rows, err := s.pool.Query(ctx, `select name, uidvalidity, messages, synced from folder where account = $1 order by name`, account)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Folder
	for rows.Next() {
		f := Folder{Account: account}
		var validity int64
		if err := rows.Scan(&f.Name, &validity, &f.Messages, &f.Synced); err != nil {
			return nil, err
		}
		f.UIDValidity, f.Synced = uint32(validity), f.Synced.UTC()
		out = append(out, f)
	}
	return out, rows.Err()
}

func (s *PgStore) Pending(ctx context.Context, limit int) ([]Event, error) {
	rows, err := s.pool.Query(ctx, `select id, key, body from outbox order by id limit $1`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Event
	for rows.Next() {
		var e Event
		var body []byte
		if err := rows.Scan(&e.ID, &e.Key, &body); err != nil {
			return nil, err
		}
		if err := json.Unmarshal(body, &e.Body); err != nil {
			return nil, err
		}
		out = append(out, e)
	}
	return out, rows.Err()
}

func (s *PgStore) Said(ctx context.Context, ids []int64) error {
	_, err := s.pool.Exec(ctx, `delete from outbox where id = any($1)`, ids)
	return err
}

func (s *PgStore) SetState(ctx context.Context, st AccountState) error {
	_, err := s.pool.Exec(ctx, `insert into account_state (account, state, why, heard, synced, said) values ($1, $2, $3, $4, $5, $6)
		on conflict (account) do update set state = excluded.state, why = excluded.why, heard = excluded.heard,
		synced = excluded.synced, said = excluded.said`,
		st.Account, st.State, st.Why, nullTime(st.Heard), nullTime(st.Synced), st.Said)
	return err
}

func nullTime(t time.Time) *time.Time {
	if t.IsZero() {
		return nil
	}
	return &t
}

func (s *PgStore) States(ctx context.Context) (map[string]AccountState, error) {
	rows, err := s.pool.Query(ctx, `select account, state, why, heard, synced, said from account_state`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[string]AccountState{}
	for rows.Next() {
		var st AccountState
		var heard, synced *time.Time
		if err := rows.Scan(&st.Account, &st.State, &st.Why, &heard, &synced, &st.Said); err != nil {
			return nil, err
		}
		if heard != nil {
			st.Heard = heard.UTC()
		}
		if synced != nil {
			st.Synced = synced.UTC()
		}
		st.Said = st.Said.UTC()
		out[st.Account] = st
	}
	return out, rows.Err()
}