Stabilize mail identity: canonical hash, index drift, HTML panic
Behebt drei Fehler derselben Klasse: an jeder Stelle wurde derselbe Wert zweimal berechnet, statt einmal berechnet und weitergereicht - und die beiden Berechnungen liefen auseinander. 1. Instabiler Ersatzschluessel (bug-hashkey-instabil.md) Mails ohne Message-ID bekamen sha256 ueber die IMAP-Rohbytes, der Reindex hashte dieselbe Mail ueber die mbox-gespeicherten Bytes (>From-Quoting, andere Zeilenenden) -> zwei Schluessel, 4.024 Doppel-Eintraege in copied. Fix: canonicalMessageBytes() bringt beide Seiten auf eine Form (>From zurueckdrehen, CRLF->LF, Trailing-Newlines weg). Alle Pfade (Index, Migration, Viewer, Dedup) nutzen dieselbe Funktion. body_sha256 in copied + mbox_index, UNIQUE erweitert. 2. Index-Drift (bug-index-drift.md) Der plain-mbox-Reader las nach Datei-Position statt nach dem gespeicherten file_offset. Weil SQLITE_BUSY (busy_timeout=0) je Ordner einen Index-Eintrag verschluckt hatte, war ab dieser Luecke alles um 1 verschoben: Klick auf Mail X zeigte Mail X+1. Betroffen genau die 6 Ordner mit Busy-Fehler beim Rettungslauf. Fix: busy_timeout=10000, Lesen ueber file_offset, --reindex --rebuild. 3. HTML-Vorschau-Panic (bug-htmltotext-panic.md) replaceCaseInsensitive/stripHTMLBlock indizierten mit Offsets aus strings.ToLower(s) in s - ToLower ist nicht byte-laengen-erhaltend (z.B. U+0130). ~0,1% der Mails brachten die Vorschau zum Absturz. Fix: asciiFoldIndex() sucht direkt auf den Original-Bytes; zusaetzlich Rohtext-Fallback, damit keine archivierte Mail unsichtbar wird. Weiter: Archiv-zuerst-Reihenfolge (mbox_done/target_done) - ein sterbendes Ziel kostet keine Archiv-Kopie mehr; Dedup vergleicht zusaetzlich den Inhalt und schuetzt byte-verschiedene Varianten. WICHTIG: Die 62.073 alten Alias-Zeilen in copied bleiben bewusst erhalten. Sie sehen wie Muell aus, sind aber die Zeilen, auf die der alte Schluessel matcht - ein Loeschen wuerde Mails erneut in fremde Postfaecher kopieren. Verifiziert: go test ./... gruen; 58.051 Archivmails, 0 ohne copied-Zeile; Drift 0 ueber alle 69 Ordner; 2.000 HTTP-Stichproben ueber 8 Ordner: 0 falsche Zuordnung, 0 Panics; Watcher 220 + 60 Mails archived=0 target=0 errors=0. Live als Image fix-identity2-20260716. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
DonVoo
parent
6c4073a01c
commit
9f5cc59af7
15 changed files with 1650 additions and 181 deletions
|
|
@ -75,7 +75,13 @@ func ConnectDB(initDB bool) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := db.Exec(`PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;`); err != nil {
|
||||
// SQLite pragmas are connection-local. Keep a single process-local
|
||||
// connection so every query uses the busy timeout configured here. The web
|
||||
// app and watcher are separate processes; WAL + busy_timeout coordinates
|
||||
// their writes without failing immediately with SQLITE_BUSY.
|
||||
db.SetMaxOpenConns(1)
|
||||
db.SetMaxIdleConns(1)
|
||||
if _, err := db.Exec(`PRAGMA busy_timeout=10000; PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;`); err != nil {
|
||||
_ = db.Close()
|
||||
return err
|
||||
}
|
||||
|
|
@ -115,14 +121,18 @@ func ConnectDB(initDB bool) error {
|
|||
account_id INTEGER NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
|
||||
folder TEXT NOT NULL,
|
||||
message_id TEXT NOT NULL,
|
||||
body_sha256 TEXT NOT NULL DEFAULT '',
|
||||
mbox_done INTEGER NOT NULL DEFAULT 0,
|
||||
target_done INTEGER NOT NULL DEFAULT 0,
|
||||
copied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
UNIQUE(account_id, folder, message_id)
|
||||
UNIQUE(account_id, folder, message_id, body_sha256)
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS mbox_index(
|
||||
account_id INTEGER NOT NULL,
|
||||
folder TEXT NOT NULL,
|
||||
seq INTEGER NOT NULL,
|
||||
message_id TEXT NOT NULL,
|
||||
body_sha256 TEXT NOT NULL DEFAULT '',
|
||||
subject TEXT NOT NULL DEFAULT '',
|
||||
from_addr TEXT NOT NULL DEFAULT '',
|
||||
date TEXT NOT NULL DEFAULT '',
|
||||
|
|
@ -130,7 +140,7 @@ func ConnectDB(initDB bool) error {
|
|||
frame_len INTEGER NOT NULL,
|
||||
inner_offset INTEGER NOT NULL DEFAULT 0,
|
||||
inner_len INTEGER NOT NULL,
|
||||
UNIQUE(account_id, folder, message_id)
|
||||
UNIQUE(account_id, folder, seq)
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS mbox_index_state(
|
||||
account_id INTEGER NOT NULL,
|
||||
|
|
@ -183,6 +193,14 @@ func ConnectDB(initDB bool) error {
|
|||
_ = db.Close()
|
||||
return err
|
||||
}
|
||||
if err := ensureCopiedStageColumns(); err != nil {
|
||||
_ = db.Close()
|
||||
return err
|
||||
}
|
||||
if err := ensureMessageIdentitySchema(); err != nil {
|
||||
_ = db.Close()
|
||||
return err
|
||||
}
|
||||
if err := seedArchiveMailboxesFromFS(); err != nil {
|
||||
_ = db.Close()
|
||||
return err
|
||||
|
|
@ -190,6 +208,153 @@ func ConnectDB(initDB bool) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// ensureMessageIdentitySchema upgrades the old Message-ID-only bookkeeping in
|
||||
// one transaction. No logical rows are discarded: old keys remain as rows with
|
||||
// an empty body hash and continue to act as compatibility aliases. The mbox
|
||||
// index is deliberately invalidated so its hashes are rebuilt from the archive
|
||||
// before the next migration pass makes a copy decision.
|
||||
func ensureMessageIdentitySchema() error {
|
||||
copiedColumns, err := tableColumns("copied")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
indexColumns, err := tableColumns("mbox_index")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
upgradeCopied := !copiedColumns["body_sha256"]
|
||||
upgradeIndex := !indexColumns["body_sha256"]
|
||||
if !upgradeCopied && !upgradeIndex {
|
||||
_, err := DB.Exec(`CREATE INDEX IF NOT EXISTS idx_copied_identity ON copied(account_id, folder, message_id, body_sha256);
|
||||
CREATE INDEX IF NOT EXISTS idx_mbox_index_identity ON mbox_index(account_id, folder, message_id, body_sha256);`)
|
||||
return err
|
||||
}
|
||||
tx, err := DB.Begin()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
if upgradeCopied {
|
||||
statements := []string{
|
||||
`ALTER TABLE copied RENAME TO copied_pre_identity`,
|
||||
`CREATE TABLE copied(
|
||||
account_id INTEGER NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
|
||||
folder TEXT NOT NULL,
|
||||
message_id TEXT NOT NULL,
|
||||
body_sha256 TEXT NOT NULL DEFAULT '',
|
||||
mbox_done INTEGER NOT NULL DEFAULT 0,
|
||||
target_done INTEGER NOT NULL DEFAULT 0,
|
||||
copied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
UNIQUE(account_id, folder, message_id, body_sha256)
|
||||
)`,
|
||||
`INSERT INTO copied(account_id, folder, message_id, body_sha256, mbox_done, target_done, copied_at)
|
||||
SELECT account_id, folder, message_id, '', mbox_done, target_done, copied_at FROM copied_pre_identity`,
|
||||
`DROP TABLE copied_pre_identity`,
|
||||
}
|
||||
for _, stmt := range statements {
|
||||
if _, err := tx.Exec(stmt); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
if upgradeIndex {
|
||||
statements := []string{
|
||||
`ALTER TABLE mbox_index RENAME TO mbox_index_pre_identity`,
|
||||
`CREATE TABLE mbox_index(
|
||||
account_id INTEGER NOT NULL,
|
||||
folder TEXT NOT NULL,
|
||||
seq INTEGER NOT NULL,
|
||||
message_id TEXT NOT NULL,
|
||||
body_sha256 TEXT NOT NULL DEFAULT '',
|
||||
subject TEXT NOT NULL DEFAULT '',
|
||||
from_addr TEXT NOT NULL DEFAULT '',
|
||||
date TEXT NOT NULL DEFAULT '',
|
||||
file_offset INTEGER NOT NULL,
|
||||
frame_len INTEGER NOT NULL,
|
||||
inner_offset INTEGER NOT NULL DEFAULT 0,
|
||||
inner_len INTEGER NOT NULL,
|
||||
UNIQUE(account_id, folder, seq)
|
||||
)`,
|
||||
`INSERT INTO mbox_index(account_id, folder, seq, message_id, body_sha256, subject, from_addr, date, file_offset, frame_len, inner_offset, inner_len)
|
||||
SELECT account_id, folder, seq, message_id, '', subject, from_addr, date, file_offset, frame_len, inner_offset, inner_len FROM mbox_index_pre_identity`,
|
||||
`DROP TABLE mbox_index_pre_identity`,
|
||||
`UPDATE mbox_index_state SET indexed_bytes=-1`,
|
||||
}
|
||||
for _, stmt := range statements {
|
||||
if _, err := tx.Exec(stmt); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, err := tx.Exec(`CREATE INDEX IF NOT EXISTS idx_copied_identity ON copied(account_id, folder, message_id, body_sha256)`); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.Exec(`CREATE INDEX IF NOT EXISTS idx_mbox_index_identity ON mbox_index(account_id, folder, message_id, body_sha256)`); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
func tableColumns(table string) (map[string]bool, error) {
|
||||
rows, err := DB.Query(`PRAGMA table_info(` + table + `)`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
columns := map[string]bool{}
|
||||
for rows.Next() {
|
||||
var cid, notNull, pk int
|
||||
var name, typ string
|
||||
var defaultValue any
|
||||
if err := rows.Scan(&cid, &name, &typ, ¬Null, &defaultValue, &pk); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
columns[name] = true
|
||||
}
|
||||
return columns, rows.Err()
|
||||
}
|
||||
|
||||
func ensureCopiedStageColumns() error {
|
||||
columns := map[string]bool{}
|
||||
rows, err := DB.Query(`PRAGMA table_info(copied)`)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for rows.Next() {
|
||||
var cid int
|
||||
var name, typ string
|
||||
var notNull int
|
||||
var defaultValue any
|
||||
var pk int
|
||||
if err := rows.Scan(&cid, &name, &typ, ¬Null, &defaultValue, &pk); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
columns[name] = true
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
// Existing copied rows were created by the old all-or-nothing pipeline and
|
||||
// therefore represent both stages as complete. DEFAULT 1 preserves that
|
||||
// truth while all new writes below set their stage values explicitly.
|
||||
if !columns["mbox_done"] {
|
||||
if _, err := DB.Exec(`ALTER TABLE copied ADD COLUMN mbox_done INTEGER NOT NULL DEFAULT 1`); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if !columns["target_done"] {
|
||||
if _, err := DB.Exec(`ALTER TABLE copied ADD COLUMN target_done INTEGER NOT NULL DEFAULT 1`); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func ensureDBFile(path string, initDB bool) error {
|
||||
info, err := os.Stat(path)
|
||||
if err == nil {
|
||||
|
|
@ -217,7 +382,6 @@ func ensureAppUserColumns() error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var cid int
|
||||
var name, typ string
|
||||
|
|
@ -225,6 +389,7 @@ func ensureAppUserColumns() error {
|
|||
var defaultValue any
|
||||
var pk int
|
||||
if err := rows.Scan(&cid, &name, &typ, ¬Null, &defaultValue, &pk); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
if name == "display_name" {
|
||||
|
|
@ -232,6 +397,10 @@ func ensureAppUserColumns() error {
|
|||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
if !hasDisplayName {
|
||||
|
|
@ -518,22 +687,129 @@ func seedArchiveMailboxesFromFS() error {
|
|||
}
|
||||
|
||||
func AlreadyCopied(accountID int64, folder, messageID string) (bool, error) {
|
||||
if messageID == "" {
|
||||
return false, nil
|
||||
}
|
||||
var x int
|
||||
err := DB.QueryRow(`SELECT 1 FROM copied WHERE account_id=? AND folder=? AND message_id=?`, accountID, folder, messageID).Scan(&x)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return false, nil
|
||||
}
|
||||
return err == nil, err
|
||||
state, err := GetCopyState(accountID, folder, messageID)
|
||||
return state.MboxDone && state.TargetDone, err
|
||||
}
|
||||
|
||||
func MarkCopied(accountID int64, folder, messageID string) error {
|
||||
if messageID == "" {
|
||||
return MarkIdentityCopied(accountID, folder, MessageIdentity{MessageID: messageID})
|
||||
}
|
||||
|
||||
type CopyState struct {
|
||||
MboxDone bool
|
||||
TargetDone bool
|
||||
}
|
||||
|
||||
func GetCopyState(accountID int64, folder, messageID string) (CopyState, error) {
|
||||
return GetCopyIdentityState(accountID, folder, MessageIdentity{MessageID: messageID})
|
||||
}
|
||||
|
||||
func MarkMboxCopied(accountID int64, folder, messageID string) error {
|
||||
return MarkIdentityMboxCopied(accountID, folder, MessageIdentity{MessageID: messageID})
|
||||
}
|
||||
|
||||
func MarkTargetCopied(accountID int64, folder, messageID string) error {
|
||||
return MarkIdentityTargetCopied(accountID, folder, MessageIdentity{MessageID: messageID})
|
||||
}
|
||||
|
||||
func GetCopyIdentityState(accountID int64, folder string, identity MessageIdentity) (CopyState, error) {
|
||||
return GetCopyIdentityStateWithFolderAlias(accountID, folder, folder, identity)
|
||||
}
|
||||
|
||||
func GetCopyIdentityStateWithFolderAlias(accountID int64, folder, legacyFolder string, identity MessageIdentity) (CopyState, error) {
|
||||
if identity.MessageID == "" && identity.BodySHA256 == "" {
|
||||
return CopyState{}, nil
|
||||
}
|
||||
folders := []string{folder}
|
||||
if legacyFolder != "" && legacyFolder != folder {
|
||||
folders = append(folders, legacyFolder)
|
||||
}
|
||||
// Precise rows always win. For Message-ID-less mail the hash is the whole
|
||||
// identity, irrespective of which legacy key happened to be stored beside it.
|
||||
var mboxDone, targetDone int
|
||||
for _, candidateFolder := range folders {
|
||||
var mbox, target int
|
||||
var err error
|
||||
if identity.MessageID == "" {
|
||||
err = DB.QueryRow(`SELECT COALESCE(MAX(mbox_done),0), COALESCE(MAX(target_done),0) FROM copied
|
||||
WHERE account_id=? AND folder=? AND body_sha256=?`, accountID, candidateFolder, identity.BodySHA256).Scan(&mbox, &target)
|
||||
} else {
|
||||
err = DB.QueryRow(`SELECT COALESCE(MAX(mbox_done),0), COALESCE(MAX(target_done),0) FROM copied
|
||||
WHERE account_id=? AND folder=? AND message_id=? AND body_sha256=?`,
|
||||
accountID, candidateFolder, identity.MessageID, identity.BodySHA256).Scan(&mbox, &target)
|
||||
}
|
||||
if err != nil {
|
||||
return CopyState{}, err
|
||||
}
|
||||
mboxDone |= mbox
|
||||
targetDone |= target
|
||||
}
|
||||
if mboxDone != 0 || targetDone != 0 {
|
||||
return CopyState{MboxDone: mboxDone != 0, TargetDone: targetDone != 0}, nil
|
||||
}
|
||||
// Once a Message-ID has precise archive identities, a different body with
|
||||
// that same ID is new and must not be hidden by the old wildcard row.
|
||||
if identity.MessageID != "" && identity.BodySHA256 != "" {
|
||||
var precise int
|
||||
for _, candidateFolder := range folders {
|
||||
var count int
|
||||
if err := DB.QueryRow(`SELECT count(*) FROM copied WHERE account_id=? AND folder=? AND message_id=? AND body_sha256<>''`,
|
||||
accountID, candidateFolder, identity.MessageID).Scan(&count); err != nil {
|
||||
return CopyState{}, err
|
||||
}
|
||||
precise += count
|
||||
}
|
||||
if precise > 0 {
|
||||
return CopyState{}, nil
|
||||
}
|
||||
}
|
||||
aliases := []string{identity.MessageID, identity.LegacyMessageID}
|
||||
if identity.MessageID == "" && identity.BodySHA256 != "" {
|
||||
aliases = append(aliases, "sha256:"+identity.BodySHA256)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, alias := range aliases {
|
||||
alias = strings.TrimSpace(alias)
|
||||
if alias == "" || seen[alias] {
|
||||
continue
|
||||
}
|
||||
seen[alias] = true
|
||||
for _, candidateFolder := range folders {
|
||||
var mbox, target int
|
||||
err := DB.QueryRow(`SELECT COALESCE(MAX(mbox_done),0), COALESCE(MAX(target_done),0) FROM copied
|
||||
WHERE account_id=? AND folder=? AND message_id=? AND body_sha256=''`, accountID, candidateFolder, alias).Scan(&mbox, &target)
|
||||
if err != nil {
|
||||
return CopyState{}, err
|
||||
}
|
||||
mboxDone |= mbox
|
||||
targetDone |= target
|
||||
}
|
||||
}
|
||||
return CopyState{MboxDone: mboxDone != 0, TargetDone: targetDone != 0}, nil
|
||||
}
|
||||
|
||||
func MarkIdentityCopied(accountID int64, folder string, identity MessageIdentity) error {
|
||||
return markCopyIdentity(accountID, folder, identity, true, true)
|
||||
}
|
||||
|
||||
func MarkIdentityMboxCopied(accountID int64, folder string, identity MessageIdentity) error {
|
||||
return markCopyIdentity(accountID, folder, identity, true, false)
|
||||
}
|
||||
|
||||
func MarkIdentityTargetCopied(accountID int64, folder string, identity MessageIdentity) error {
|
||||
return markCopyIdentity(accountID, folder, identity, false, true)
|
||||
}
|
||||
|
||||
func markCopyIdentity(accountID int64, folder string, identity MessageIdentity, mboxDone, targetDone bool) error {
|
||||
if identity.MessageID == "" && identity.BodySHA256 == "" {
|
||||
return nil
|
||||
}
|
||||
_, err := DB.Exec(`INSERT OR IGNORE INTO copied(account_id, folder, message_id) VALUES(?,?,?)`, accountID, folder, messageID)
|
||||
_, err := DB.Exec(`INSERT INTO copied(account_id, folder, message_id, body_sha256, mbox_done, target_done)
|
||||
VALUES(?,?,?,?,?,?)
|
||||
ON CONFLICT(account_id, folder, message_id, body_sha256) DO UPDATE SET
|
||||
mbox_done=MAX(copied.mbox_done, excluded.mbox_done),
|
||||
target_done=MAX(copied.target_done, excluded.target_done),
|
||||
copied_at=CURRENT_TIMESTAMP`, accountID, folder, identity.MessageID, identity.BodySHA256, boolInt(mboxDone), boolInt(targetDone))
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
@ -542,6 +818,7 @@ type MboxIndexEntry struct {
|
|||
Folder string
|
||||
Seq int
|
||||
MessageID string
|
||||
BodySHA256 string
|
||||
Subject string
|
||||
From string
|
||||
Date string
|
||||
|
|
@ -552,7 +829,7 @@ type MboxIndexEntry struct {
|
|||
}
|
||||
|
||||
func SaveMboxIndex(e MboxIndexEntry) error {
|
||||
if DB == nil || e.MessageID == "" {
|
||||
if DB == nil || (e.MessageID == "" && e.BodySHA256 == "") {
|
||||
return nil
|
||||
}
|
||||
var seq int
|
||||
|
|
@ -561,9 +838,9 @@ func SaveMboxIndex(e MboxIndexEntry) error {
|
|||
} else {
|
||||
_ = DB.QueryRow(`SELECT COALESCE(MAX(seq), -1) + 1 FROM mbox_index WHERE account_id=? AND folder=?`, e.AccountID, e.Folder).Scan(&seq)
|
||||
}
|
||||
_, err := DB.Exec(`INSERT OR IGNORE INTO mbox_index(account_id, folder, seq, message_id, subject, from_addr, date, file_offset, frame_len, inner_offset, inner_len)
|
||||
VALUES(?,?,?,?,?,?,?,?,?,?,?)`,
|
||||
e.AccountID, e.Folder, seq, e.MessageID, e.Subject, e.From, e.Date, e.FileOffset, e.FrameLen, e.InnerOffset, e.InnerLen)
|
||||
_, err := DB.Exec(`INSERT OR REPLACE INTO mbox_index(account_id, folder, seq, message_id, body_sha256, subject, from_addr, date, file_offset, frame_len, inner_offset, inner_len)
|
||||
VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`,
|
||||
e.AccountID, e.Folder, seq, e.MessageID, e.BodySHA256, e.Subject, e.From, e.Date, e.FileOffset, e.FrameLen, e.InnerOffset, e.InnerLen)
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
@ -576,6 +853,53 @@ func ReplaceMboxIndex(accountID int64, folder string, entries []MboxIndexEntry,
|
|||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
folderAliases := []string{folder}
|
||||
aliasRows, err := tx.Query(`SELECT folder FROM copied WHERE account_id=? UNION SELECT folder FROM mbox_index WHERE account_id=?`, accountID, accountID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for aliasRows.Next() {
|
||||
var candidate string
|
||||
if err := aliasRows.Scan(&candidate); err != nil {
|
||||
_ = aliasRows.Close()
|
||||
return err
|
||||
}
|
||||
if candidate != folder && safeMboxName(candidate) == folder {
|
||||
folderAliases = append(folderAliases, candidate)
|
||||
}
|
||||
}
|
||||
if err := aliasRows.Err(); err != nil {
|
||||
_ = aliasRows.Close()
|
||||
return err
|
||||
}
|
||||
if err := aliasRows.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
legacyIDsBySeq := map[int]string{}
|
||||
for _, candidateFolder := range folderAliases {
|
||||
rows, err := tx.Query(`SELECT seq, message_id FROM mbox_index WHERE account_id=? AND folder=?`, accountID, candidateFolder)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for rows.Next() {
|
||||
var seq int
|
||||
var messageID string
|
||||
if err := rows.Scan(&seq, &messageID); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
if _, exists := legacyIDsBySeq[seq]; !exists || candidateFolder == folder {
|
||||
legacyIDsBySeq[seq] = messageID
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if _, err := tx.Exec(`DELETE FROM mbox_index WHERE account_id=? AND folder=?`, accountID, folder); err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -583,15 +907,33 @@ func ReplaceMboxIndex(accountID int64, folder string, entries []MboxIndexEntry,
|
|||
entry.AccountID = accountID
|
||||
entry.Folder = folder
|
||||
entry.Seq = i
|
||||
if entry.MessageID == "" {
|
||||
if entry.MessageID == "" && entry.BodySHA256 == "" {
|
||||
continue
|
||||
}
|
||||
if _, err := tx.Exec(`INSERT INTO mbox_index(account_id, folder, seq, message_id, subject, from_addr, date, file_offset, frame_len, inner_offset, inner_len)
|
||||
VALUES(?,?,?,?,?,?,?,?,?,?,?)`,
|
||||
entry.AccountID, entry.Folder, entry.Seq, entry.MessageID, entry.Subject, entry.From, entry.Date,
|
||||
if _, err := tx.Exec(`INSERT INTO mbox_index(account_id, folder, seq, message_id, body_sha256, subject, from_addr, date, file_offset, frame_len, inner_offset, inner_len)
|
||||
VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`,
|
||||
entry.AccountID, entry.Folder, entry.Seq, entry.MessageID, entry.BodySHA256, entry.Subject, entry.From, entry.Date,
|
||||
entry.FileOffset, entry.FrameLen, entry.InnerOffset, entry.InnerLen); err != nil {
|
||||
return err
|
||||
}
|
||||
// A full rebuild recovers records which reached the old append-first
|
||||
// pipeline but whose index/copied writes lost a SQLITE_BUSY race. In that
|
||||
// pipeline the target append happened before the mbox append, so recovered
|
||||
// records are complete in both stages. Marking them prevents a later watch
|
||||
// run from appending duplicates to either destination.
|
||||
state, found, err := copyStateForReindex(tx, entry.AccountID, folderAliases, entry, legacyIDsBySeq[i])
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !found {
|
||||
state = CopyState{MboxDone: true, TargetDone: true}
|
||||
}
|
||||
if _, err := tx.Exec(`INSERT INTO copied(account_id, folder, message_id, body_sha256, mbox_done, target_done)
|
||||
VALUES(?,?,?,?,?,?)
|
||||
ON CONFLICT(account_id, folder, message_id, body_sha256) DO UPDATE SET mbox_done=1`,
|
||||
entry.AccountID, entry.Folder, entry.MessageID, entry.BodySHA256, 1, boolInt(state.TargetDone)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if _, err := tx.Exec(`INSERT INTO mbox_index_state(account_id, folder, indexed_bytes)
|
||||
VALUES(?,?,?)
|
||||
|
|
@ -602,6 +944,88 @@ func ReplaceMboxIndex(accountID int64, folder string, entries []MboxIndexEntry,
|
|||
return tx.Commit()
|
||||
}
|
||||
|
||||
// RemoveMboxIndexFolderAliases removes only superseded index metadata such as
|
||||
// "INBOX.Newbies " after the actual archive file has been rebuilt under its
|
||||
// canonical filesystem name "INBOX.Newbies". copied compatibility aliases and
|
||||
// all mbox files remain untouched.
|
||||
func RemoveMboxIndexFolderAliases(accountID int64, canonicalFolders map[string]bool) error {
|
||||
rows, err := DB.Query(`SELECT folder FROM mbox_index WHERE account_id=? UNION SELECT folder FROM mbox_index_state WHERE account_id=?`, accountID, accountID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var stale []string
|
||||
for rows.Next() {
|
||||
var folder string
|
||||
if err := rows.Scan(&folder); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
if !canonicalFolders[folder] && canonicalFolders[safeMboxName(folder)] {
|
||||
stale = append(stale, folder)
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
if len(stale) == 0 {
|
||||
return nil
|
||||
}
|
||||
tx, err := DB.Begin()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
for _, folder := range stale {
|
||||
if _, err := tx.Exec(`DELETE FROM mbox_index WHERE account_id=? AND folder=?`, accountID, folder); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.Exec(`DELETE FROM mbox_index_state WHERE account_id=? AND folder=?`, accountID, folder); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
func copyStateForReindex(tx *sql.Tx, accountID int64, folders []string, entry MboxIndexEntry, legacyMessageID string) (CopyState, bool, error) {
|
||||
type candidate struct{ messageID, bodyHash string }
|
||||
candidates := []candidate{{entry.MessageID, entry.BodySHA256}}
|
||||
if legacyMessageID != "" {
|
||||
candidates = append(candidates, candidate{legacyMessageID, ""})
|
||||
}
|
||||
if entry.MessageID != "" {
|
||||
candidates = append(candidates, candidate{entry.MessageID, ""})
|
||||
} else if entry.BodySHA256 != "" {
|
||||
candidates = append(candidates, candidate{"sha256:" + entry.BodySHA256, ""})
|
||||
}
|
||||
seen := map[candidate]bool{}
|
||||
var state CopyState
|
||||
found := false
|
||||
for _, c := range candidates {
|
||||
if seen[c] {
|
||||
continue
|
||||
}
|
||||
seen[c] = true
|
||||
for _, folder := range folders {
|
||||
var count, mboxDone, targetDone int
|
||||
if err := tx.QueryRow(`SELECT count(*), COALESCE(MAX(mbox_done),0), COALESCE(MAX(target_done),0)
|
||||
FROM copied WHERE account_id=? AND folder=? AND message_id=? AND body_sha256=?`,
|
||||
accountID, folder, c.messageID, c.bodyHash).Scan(&count, &mboxDone, &targetDone); err != nil {
|
||||
return CopyState{}, false, err
|
||||
}
|
||||
if count > 0 {
|
||||
found = true
|
||||
state.MboxDone = state.MboxDone || mboxDone != 0
|
||||
state.TargetDone = state.TargetDone || targetDone != 0
|
||||
}
|
||||
}
|
||||
}
|
||||
return state, found, nil
|
||||
}
|
||||
|
||||
func UpdateMboxIndexState(accountID int64, folder string, indexedBytes int64) error {
|
||||
if DB == nil || accountID == 0 {
|
||||
return nil
|
||||
|
|
@ -629,7 +1053,7 @@ func ListMboxIndex(accountID int64, folder string) ([]MboxIndexEntry, error) {
|
|||
if DB == nil || accountID == 0 {
|
||||
return nil, sql.ErrNoRows
|
||||
}
|
||||
rows, err := DB.Query(`SELECT account_id, folder, seq, message_id, subject, from_addr, date, file_offset, frame_len, inner_offset, inner_len
|
||||
rows, err := DB.Query(`SELECT account_id, folder, seq, message_id, body_sha256, subject, from_addr, date, file_offset, frame_len, inner_offset, inner_len
|
||||
FROM mbox_index WHERE account_id=? AND folder=? ORDER BY seq`, accountID, folder)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
|
@ -638,7 +1062,7 @@ func ListMboxIndex(accountID int64, folder string) ([]MboxIndexEntry, error) {
|
|||
var out []MboxIndexEntry
|
||||
for rows.Next() {
|
||||
var e MboxIndexEntry
|
||||
if err := rows.Scan(&e.AccountID, &e.Folder, &e.Seq, &e.MessageID, &e.Subject, &e.From, &e.Date, &e.FileOffset, &e.FrameLen, &e.InnerOffset, &e.InnerLen); err != nil {
|
||||
if err := rows.Scan(&e.AccountID, &e.Folder, &e.Seq, &e.MessageID, &e.BodySHA256, &e.Subject, &e.From, &e.Date, &e.FileOffset, &e.FrameLen, &e.InnerOffset, &e.InnerLen); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, e)
|
||||
|
|
@ -650,10 +1074,10 @@ func GetMboxIndex(accountID int64, folder string, seq int) (MboxIndexEntry, erro
|
|||
if DB == nil || accountID == 0 {
|
||||
return MboxIndexEntry{}, sql.ErrNoRows
|
||||
}
|
||||
row := DB.QueryRow(`SELECT account_id, folder, seq, message_id, subject, from_addr, date, file_offset, frame_len, inner_offset, inner_len
|
||||
row := DB.QueryRow(`SELECT account_id, folder, seq, message_id, body_sha256, subject, from_addr, date, file_offset, frame_len, inner_offset, inner_len
|
||||
FROM mbox_index WHERE account_id=? AND folder=? AND seq=?`, accountID, folder, seq)
|
||||
var e MboxIndexEntry
|
||||
err := row.Scan(&e.AccountID, &e.Folder, &e.Seq, &e.MessageID, &e.Subject, &e.From, &e.Date, &e.FileOffset, &e.FrameLen, &e.InnerOffset, &e.InnerLen)
|
||||
err := row.Scan(&e.AccountID, &e.Folder, &e.Seq, &e.MessageID, &e.BodySHA256, &e.Subject, &e.From, &e.Date, &e.FileOffset, &e.FrameLen, &e.InnerOffset, &e.InnerLen)
|
||||
return e, err
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue