1package db
  2
  3import (
  4	"context"
  5	"database/sql"
  6	"errors"
  7	"fmt"
  8	"log/slog"
  9
 10	_ "modernc.org/sqlite"
 11)
 12
 13var schemaVersion = len(schemaChanges())
 14
 15const pragmas = `
 16PRAGMA journal_mode  = WAL;
 17PRAGMA synchronous   = NORMAL;
 18PRAGMA foreign_keys  = ON;
 19PRAGMA busy_timeout  = 5000;
 20PRAGMA temp_store    = MEMORY;
 21`
 22
 23func schemaChanges() []string {
 24
 25	schema := []string{`
 26
 27CREATE TABLE IF NOT EXISTS users (
 28    id            INTEGER PRIMARY KEY,
 29    name          TEXT    NOT NULL UNIQUE,
 30    password_hash BLOB    NOT NULL,
 31    created_at    INTEGER NOT NULL
 32) STRICT;
 33
 34CREATE INDEX IF NOT EXISTS idx_users_name ON users(name);
 35
 36CREATE TABLE IF NOT EXISTS addresses (
 37    id            INTEGER PRIMARY KEY,
 38    user_id       INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
 39    name          TEXT    NOT NULL UNIQUE,
 40    created_at    INTEGER NOT NULL
 41) STRICT;
 42
 43CREATE INDEX IF NOT EXISTS idx_addresses_user_id ON addresses(user_id);
 44CREATE INDEX IF NOT EXISTS idx_addresses_name ON addresses(name);
 45
 46CREATE TABLE IF NOT EXISTS gatekeepers (
 47    id            INTEGER PRIMARY KEY,
 48    user_id       INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
 49    from_address  TEXT    NOT NULL,
 50    destination   TEXT    NOT NULL,
 51    created_at    INTEGER NOT NULL,
 52    UNIQUE (user_id, from_address)
 53) STRICT;
 54
 55CREATE TABLE IF NOT EXISTS mailboxes (
 56    id             INTEGER PRIMARY KEY AUTOINCREMENT,
 57    user_id        INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
 58    name           TEXT    NOT NULL,
 59    special_use    TEXT,
 60    uidvalidity    INTEGER NOT NULL,
 61    uidnext        INTEGER NOT NULL DEFAULT 1,
 62    highest_modseq INTEGER NOT NULL DEFAULT 0,
 63    UNIQUE (user_id, name)
 64) STRICT;
 65
 66CREATE TABLE IF NOT EXISTS subscriptions (
 67    id             INTEGER PRIMARY KEY,
 68    user_id        INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
 69    mailbox_id     INTEGER REFERENCES mailboxes(id) ON DELETE SET NULL,
 70    mailbox_name   TEXT, -- mailboxes can be subscribed even if mailbox does not exist, hence duplicate name
 71    UNIQUE (user_id, mailbox_id)
 72) STRICT;
 73
 74CREATE TABLE IF NOT EXISTS messages (
 75    id            INTEGER PRIMARY KEY,
 76    blob_address  TEXT    NOT NULL UNIQUE,
 77    size          INTEGER NOT NULL,
 78    subject       TEXT,
 79    message_id    TEXT,
 80    in_reply_to   TEXT,
 81
 82    -- Envelope fields
 83    date          TEXT,
 84    from_addr     TEXT,             -- JSON array of addresses
 85    sender        TEXT,             -- JSON array of addresses
 86    reply_to      TEXT,             -- JSON array of addresses
 87    to_addr       TEXT,             -- JSON array of addresses
 88    cc            TEXT,             -- JSON array of addresses
 89    bcc           TEXT              -- JSON array of addresses
 90) STRICT;
 91
 92CREATE TABLE IF NOT EXISTS mailbox_messages (
 93    mailbox_id      INTEGER NOT NULL REFERENCES mailboxes(id) ON DELETE CASCADE,
 94    uid             INTEGER NOT NULL,
 95    message_id      INTEGER NOT NULL REFERENCES messages(id),
 96    modseq          INTEGER NOT NULL,
 97    internal_date   TEXT NOT NULL,
 98    expunged_modseq INTEGER,
 99    PRIMARY KEY (mailbox_id, uid)
100) STRICT;
101
102CREATE TABLE IF NOT EXISTS message_flags (
103    mailbox_id INTEGER NOT NULL,
104    uid        INTEGER NOT NULL,
105    flag       TEXT    NOT NULL,
106    PRIMARY KEY (mailbox_id, uid, flag),
107    FOREIGN KEY (mailbox_id) REFERENCES mailboxes(id) ON DELETE CASCADE,
108    FOREIGN KEY (mailbox_id, uid) REFERENCES mailbox_messages(mailbox_id, uid) ON DELETE CASCADE ON UPDATE CASCADE
109) STRICT;
110
111CREATE TABLE IF NOT EXISTS dkim (
112    id             INTEGER PRIMARY KEY,
113    created_at     INTEGER NOT NULL,
114    domain         TEXT NOT NULL,
115    key_type       TEXT NOT NULL,
116    blob_address   TEXT NOT NULL,
117    selector       TEXT NOT NULL,
118    enabled        INTEGER NOT NULL,
119    UNIQUE (domain, selector)
120) STRICT;
121
122CREATE TABLE IF NOT EXISTS smtp_auth (
123    id            INTEGER PRIMARY KEY,
124    name          TEXT    NOT NULL UNIQUE,
125    password_hash BLOB    NOT NULL,
126    created_at    INTEGER NOT NULL
127) STRICT;
128
129CREATE INDEX IF NOT EXISTS idx_mailbox_messages_modseq
130    ON mailbox_messages(mailbox_id, modseq);
131
132CREATE INDEX IF NOT EXISTS idx_mailbox_messages_message
133    ON mailbox_messages(message_id);
134
135CREATE TABLE IF NOT EXISTS keywords (
136    message_id INTEGER NOT NULL REFERENCES messages(id),
137    keyword    TEXT    NOT NULL,
138    PRIMARY KEY (message_id, keyword)
139) STRICT;
140
141CREATE INDEX IF NOT EXISTS idx_keywords_keyword
142    ON keywords(keyword);
143
144CREATE TABLE IF NOT EXISTS uidvalidity_counter (
145    id      INTEGER PRIMARY KEY CHECK (id = 1),
146    counter INTEGER NOT NULL DEFAULT 0
147) STRICT;
148
149INSERT INTO uidvalidity_counter (id, counter)
150VALUES (1, 0)
151ON CONFLICT(id) DO NOTHING;
152
153CREATE TABLE IF NOT EXISTS schema_version (
154    version INT NOT NULL
155) STRICT;
156`}
157
158	schema = append(schema, `
159CREATE TABLE kv_store (
160    key        TEXT PRIMARY KEY,
161    value      ANY,
162    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
163) WITHOUT ROWID; -- WITHOUT ROWID optimizes key-value lookups
164`)
165
166	return schema
167}
168
169func (db *DB) applyPragmas(ctx context.Context) error {
170	for _, conn := range []*sql.DB{db.write, db.read} {
171		if _, err := conn.ExecContext(ctx, pragmas); err != nil {
172			return fmt.Errorf("applying pragmas: %w", err)
173		}
174	}
175	return nil
176}
177
178func (db *DB) applySchema(ctx context.Context) error {
179	var currentVersion int
180	// Ignore the error on purpose in case schema_version does not exist _yet_.
181	// The very first schema can be applied idempotently.
182	_ = db.read.QueryRowContext(ctx, `SELECT version FROM schema_version`).Scan(&currentVersion)
183
184	for version := currentVersion; version < schemaVersion; version++ {
185		newVersion := version + 1
186
187		slog.InfoContext(ctx, "updating schema version", "from", currentVersion, "to", newVersion)
188
189		tx, err := db.GetWriteTx(ctx)
190		if err != nil {
191			return errors.Join(fmt.Errorf("migration %d transaction not started", newVersion), err)
192		}
193
194		if _, err := tx.ExecContext(ctx, schemaChanges()[version]); err != nil {
195			errRollback := tx.Rollback()
196			return errors.Join(fmt.Errorf("migration %d rolled back", newVersion), err, errRollback)
197		}
198
199		if _, err := tx.ExecContext(ctx, `DELETE FROM schema_version`); err != nil {
200			errRollback := tx.Rollback()
201			return errors.Join(fmt.Errorf("migration %d rolled back", newVersion), err, errRollback)
202		}
203
204		if _, err := tx.ExecContext(ctx, `INSERT INTO schema_version (version) VALUES (?)`, newVersion); err != nil {
205			errRollback := tx.Rollback()
206			return errors.Join(fmt.Errorf("migration %d rolled back", newVersion), err, errRollback)
207		}
208
209		if err := tx.Commit(); err != nil {
210			return errors.Join(fmt.Errorf("migration %d not commited", newVersion), err)
211		}
212	}
213
214	slog.InfoContext(ctx, "applying schema", "version", schemaVersion)
215	return nil
216}
217
218func (db *DB) Close() error {
219	werr := db.write.Close()
220	rerr := db.read.Close()
221	if werr != nil {
222		return werr
223	}
224	return rerr
225}
226
227func txRollback(tx *sql.Tx) {
228	err := tx.Rollback()
229	if !errors.Is(err, sql.ErrTxDone) {
230		slog.Error("failed to roll back transaction", "error", err)
231	}
232}