Fix archive reindex and add target dedup
This commit is contained in:
parent
e612896e06
commit
c83225f5ad
9 changed files with 796 additions and 6 deletions
16
INSTALL.md
16
INSTALL.md
|
|
@ -50,13 +50,25 @@ Ohne Web, z. B. als Cronjob oder Dauerlauf im Umzugsfenster:
|
|||
Dank Message-ID-Cache kopiert jeder Lauf nur Neues — beliebig oft wiederholbar
|
||||
ohne Dubletten. Für den Dauerlauf reicht `nohup`/systemd/Aufgabenplanung.
|
||||
|
||||
## 5. Sicherheit
|
||||
## 5. Wartung
|
||||
|
||||
```sh
|
||||
./mail-graveyard --reindex all
|
||||
./mail-graveyard --dedup-target konto-name # Dry-Run
|
||||
./mail-graveyard --dedup-target konto-name --apply # loescht Dubletten im Ziel
|
||||
```
|
||||
|
||||
`--reindex` baut den Archiv-Index fuer vorhandene mbox/mbox.zst-Dateien neu auf.
|
||||
`--dedup-target` ist bewusst ein Dry-Run, bis `--apply` gesetzt wird; Mails ohne
|
||||
Message-ID werden nie geloescht.
|
||||
|
||||
## 6. Sicherheit
|
||||
|
||||
Tool hält fremde IMAP-Passwörter und kann Mails senden → Default bind
|
||||
`127.0.0.1`. Falls remote nötig: hinter Caddy + `/vadmin`-Mail-2FA. `config.json`
|
||||
und `*.db` nie committen.
|
||||
|
||||
## 6. Reihenfolge bei einem echten Umzug
|
||||
## 7. Reihenfolge bei einem echten Umzug
|
||||
|
||||
1. Ziel-Postfächer beim neuen Hoster anlegen (`archiv-…`, `temp-…`).
|
||||
2. Konten im Browser eintragen, „Test" grün.
|
||||
|
|
|
|||
|
|
@ -132,6 +132,12 @@ func ConnectDB(initDB bool) error {
|
|||
inner_len INTEGER NOT NULL,
|
||||
UNIQUE(account_id, folder, message_id)
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS mbox_index_state(
|
||||
account_id INTEGER NOT NULL,
|
||||
folder TEXT NOT NULL,
|
||||
indexed_bytes INTEGER NOT NULL DEFAULT 0,
|
||||
UNIQUE(account_id, folder)
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS jobs(
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
account_id INTEGER NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
|
||||
|
|
@ -442,6 +448,23 @@ func DeleteAccount(name string) error {
|
|||
return err
|
||||
}
|
||||
|
||||
func FolderMap(accountID int64) (map[string]string, error) {
|
||||
rows, err := DB.Query(`SELECT src_folder, dst_folder FROM folder_map WHERE account_id=?`, accountID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[string]string{}
|
||||
for rows.Next() {
|
||||
var src, dst string
|
||||
if err := rows.Scan(&src, &dst); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[src] = dst
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func ListArchiveMailboxes() ([]string, error) {
|
||||
rows, err := DB.Query(`SELECT name FROM archive_mailboxes ORDER BY lower(name), name`)
|
||||
if err != nil {
|
||||
|
|
@ -544,6 +567,64 @@ func SaveMboxIndex(e MboxIndexEntry) error {
|
|||
return err
|
||||
}
|
||||
|
||||
func ReplaceMboxIndex(accountID int64, folder string, entries []MboxIndexEntry, indexedBytes int64) error {
|
||||
if DB == nil || accountID == 0 {
|
||||
return nil
|
||||
}
|
||||
tx, err := DB.Begin()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
if _, err := tx.Exec(`DELETE FROM mbox_index WHERE account_id=? AND folder=?`, accountID, folder); err != nil {
|
||||
return err
|
||||
}
|
||||
for i, entry := range entries {
|
||||
entry.AccountID = accountID
|
||||
entry.Folder = folder
|
||||
entry.Seq = i
|
||||
if entry.MessageID == "" {
|
||||
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,
|
||||
entry.FileOffset, entry.FrameLen, entry.InnerOffset, entry.InnerLen); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if _, err := tx.Exec(`INSERT INTO mbox_index_state(account_id, folder, indexed_bytes)
|
||||
VALUES(?,?,?)
|
||||
ON CONFLICT(account_id, folder) DO UPDATE SET indexed_bytes=excluded.indexed_bytes`,
|
||||
accountID, folder, indexedBytes); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
func UpdateMboxIndexState(accountID int64, folder string, indexedBytes int64) error {
|
||||
if DB == nil || accountID == 0 {
|
||||
return nil
|
||||
}
|
||||
_, err := DB.Exec(`INSERT INTO mbox_index_state(account_id, folder, indexed_bytes)
|
||||
VALUES(?,?,?)
|
||||
ON CONFLICT(account_id, folder) DO UPDATE SET indexed_bytes=excluded.indexed_bytes`,
|
||||
accountID, folder, indexedBytes)
|
||||
return err
|
||||
}
|
||||
|
||||
func MboxIndexedBytes(accountID int64, folder string) (int64, error) {
|
||||
if DB == nil || accountID == 0 {
|
||||
return 0, sql.ErrNoRows
|
||||
}
|
||||
var n int64
|
||||
err := DB.QueryRow(`SELECT indexed_bytes FROM mbox_index_state WHERE account_id=? AND folder=?`, accountID, folder).Scan(&n)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return 0, nil
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
|
||||
func ListMboxIndex(accountID int64, folder string) ([]MboxIndexEntry, error) {
|
||||
if DB == nil || accountID == 0 {
|
||||
return nil, sql.ErrNoRows
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ type TargetMailbox interface {
|
|||
Append(folder string, m RawMessage) error // mit m.Flags und m.InternalDate
|
||||
Headers(folder string, limit, offset int) ([]MessageHeader, error)
|
||||
FetchOne(folder string, uid uint32) (RawMessage, error)
|
||||
DeleteUIDs(folder string, uids []uint32) error
|
||||
Close() error
|
||||
}
|
||||
|
||||
|
|
@ -129,6 +130,32 @@ func (t *imapTarget) FetchOne(folder string, uid uint32) (RawMessage, error) {
|
|||
return t.mailbox.fetchOne(folder, uid)
|
||||
}
|
||||
|
||||
func (t *imapTarget) DeleteUIDs(folder string, uids []uint32) error {
|
||||
if len(uids) == 0 {
|
||||
return nil
|
||||
}
|
||||
if _, err := t.mailbox.c.Select(folder, nil).Wait(); err != nil {
|
||||
return err
|
||||
}
|
||||
uidSet := imap.UIDSetNum(uint32sToUIDs(uids)...)
|
||||
flags := &imap.StoreFlags{Op: imap.StoreFlagsAdd, Silent: true, Flags: []imap.Flag{imap.FlagDeleted}}
|
||||
if err := t.mailbox.c.Store(uidSet, flags, nil).Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err := t.mailbox.c.UIDExpunge(uidSet).Collect()
|
||||
return err
|
||||
}
|
||||
|
||||
func (t *imapTarget) Close() error {
|
||||
return t.mailbox.Close()
|
||||
}
|
||||
|
||||
func uint32sToUIDs(values []uint32) []imap.UID {
|
||||
out := make([]imap.UID, 0, len(values))
|
||||
for _, value := range values {
|
||||
if value != 0 {
|
||||
out = append(out, imap.UID(value))
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
|
|
|||
|
|
@ -114,6 +114,11 @@ func mboxRecord(m RawMessage) []byte {
|
|||
|
||||
func mboxAppendInfoFromMessage(m RawMessage, record []byte) MboxAppendInfo {
|
||||
info := MboxAppendInfo{InnerLen: int64(len(record))}
|
||||
parts := splitMboxRecordsWithOffsets(record)
|
||||
if len(parts) > 0 {
|
||||
info.InnerOffset = parts[0].Offset
|
||||
info.InnerLen = int64(len(parts[0].Message))
|
||||
}
|
||||
msg, err := mail.ReadMessage(bytes.NewReader(m.Body))
|
||||
if err != nil {
|
||||
return info
|
||||
|
|
@ -137,6 +142,9 @@ type MboxEntry struct {
|
|||
// ReadMboxList parst die Kopfzeilen aller Mails einer mbox-Datei (fuer die
|
||||
// Nachrichtenliste). ReadMboxMessage liefert eine einzelne Mail als Rohtext.
|
||||
func ReadMboxList(path string) ([]MboxEntry, error) {
|
||||
if err := EnsureMboxIndexed(path); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if entries, ok := readMboxListFromIndex(path); ok {
|
||||
return entries, nil
|
||||
}
|
||||
|
|
@ -162,6 +170,9 @@ func ReadMboxList(path string) ([]MboxEntry, error) {
|
|||
}
|
||||
|
||||
func ReadMboxMessage(path string, index int) ([]byte, error) {
|
||||
if err := EnsureMboxIndexed(path); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if raw, ok, err := readMboxMessageFromIndex(path, index); ok || err != nil {
|
||||
return raw, err
|
||||
}
|
||||
|
|
@ -194,6 +205,229 @@ func readMboxMessages(path string) ([][]byte, error) {
|
|||
return readMboxMessagesBytes(b), nil
|
||||
}
|
||||
|
||||
func EnsureMboxIndexed(path string) error {
|
||||
accountID, folder, ok := mboxIndexContext(path)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
indexed, err := MboxIndexedBytes(accountID, folder)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if indexed == info.Size() {
|
||||
return nil
|
||||
}
|
||||
return ReindexMboxFile(accountID, folder, path)
|
||||
}
|
||||
|
||||
func ReindexMboxFile(accountID int64, folder, path string) error {
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var entries []MboxIndexEntry
|
||||
if strings.HasSuffix(strings.ToLower(path), ".zst") {
|
||||
entries, err = reindexZstdMbox(path)
|
||||
} else {
|
||||
entries, err = reindexPlainMbox(path)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return ReplaceMboxIndex(accountID, folder, entries, info.Size())
|
||||
}
|
||||
|
||||
func ReindexArchives(name string) error {
|
||||
name = strings.TrimSpace(name)
|
||||
if name == "" {
|
||||
return fmt.Errorf("reindex ziel fehlt")
|
||||
}
|
||||
accounts, err := ListAccounts()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
matched := 0
|
||||
for _, account := range accounts {
|
||||
archive := accountMboxDir(account)
|
||||
if archive == "" {
|
||||
continue
|
||||
}
|
||||
if name != "all" && !strings.EqualFold(name, account.Name) && !strings.EqualFold(name, archive) {
|
||||
continue
|
||||
}
|
||||
if err := reindexAccountArchive(account); err != nil {
|
||||
return err
|
||||
}
|
||||
matched++
|
||||
}
|
||||
if matched == 0 {
|
||||
return fmt.Errorf("kein Konto/Archiv fuer reindex %q gefunden", name)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func reindexAccountArchive(account Account) error {
|
||||
dir := filepath.Join(Cfg.MboxRoot, accountMboxDir(account))
|
||||
for _, path := range archiveMboxPaths(dir) {
|
||||
folder := folderNameFromMboxPath(path)
|
||||
if folder == "" {
|
||||
continue
|
||||
}
|
||||
if err := ReindexMboxFile(account.ID, folder, path); err != nil {
|
||||
return fmt.Errorf("%s %s: %w", account.Name, folder, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func reindexPlainMbox(path string) ([]MboxIndexEntry, error) {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []MboxIndexEntry
|
||||
for _, part := range splitMboxRecordsWithOffsets(b) {
|
||||
entry := indexEntryFromMboxRecord(part.Message, part.Offset, int64(part.Length), 0, int64(part.Length))
|
||||
if entry.MessageID != "" {
|
||||
out = append(out, entry)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func reindexZstdMbox(path string) ([]MboxIndexEntry, error) {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dec, err := zstd.NewReader(nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer dec.Close()
|
||||
var out []MboxIndexEntry
|
||||
for _, frame := range splitZstdFrames(b) {
|
||||
record, err := dec.DecodeAll(frame.Data, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, part := range splitMboxRecordsWithOffsets(record) {
|
||||
entry := indexEntryFromMboxRecord(part.Message, frame.Offset, int64(frame.Length), part.Offset, int64(len(part.Message)))
|
||||
if entry.MessageID != "" {
|
||||
out = append(out, entry)
|
||||
}
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
type mboxRecordPart struct {
|
||||
Offset int64
|
||||
Length int
|
||||
Message []byte
|
||||
}
|
||||
|
||||
func splitMboxRecordsWithOffsets(b []byte) []mboxRecordPart {
|
||||
normalized := bytes.ReplaceAll(b, []byte("\r\n"), []byte("\n"))
|
||||
lines := bytes.SplitAfter(normalized, []byte("\n"))
|
||||
var out []mboxRecordPart
|
||||
var cur bytes.Buffer
|
||||
var curStart int64
|
||||
var pos int64
|
||||
inMsg := false
|
||||
for _, lineWithNL := range lines {
|
||||
line := bytes.TrimSuffix(lineWithNL, []byte("\n"))
|
||||
if bytes.HasPrefix(line, []byte("From ")) {
|
||||
if inMsg && cur.Len() > 0 {
|
||||
msg := bytes.TrimRight(cur.Bytes(), "\n")
|
||||
out = append(out, mboxRecordPart{Offset: curStart, Length: int(pos - curStart), Message: append([]byte(nil), msg...)})
|
||||
cur.Reset()
|
||||
}
|
||||
inMsg = true
|
||||
pos += int64(len(lineWithNL))
|
||||
curStart = pos
|
||||
continue
|
||||
}
|
||||
if inMsg {
|
||||
if bytes.HasPrefix(line, []byte(">From ")) {
|
||||
line = line[1:]
|
||||
}
|
||||
_, _ = cur.Write(line)
|
||||
_ = cur.WriteByte('\n')
|
||||
}
|
||||
pos += int64(len(lineWithNL))
|
||||
}
|
||||
if inMsg && cur.Len() > 0 {
|
||||
msg := bytes.TrimRight(cur.Bytes(), "\n")
|
||||
out = append(out, mboxRecordPart{Offset: curStart, Length: int(pos - curStart), Message: append([]byte(nil), msg...)})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
type zstdFramePart struct {
|
||||
Offset int64
|
||||
Length int
|
||||
Data []byte
|
||||
}
|
||||
|
||||
func splitZstdFrames(b []byte) []zstdFramePart {
|
||||
magic := []byte{0x28, 0xb5, 0x2f, 0xfd}
|
||||
var starts []int
|
||||
for i := 0; i <= len(b)-len(magic); i++ {
|
||||
if bytes.Equal(b[i:i+len(magic)], magic) {
|
||||
starts = append(starts, i)
|
||||
}
|
||||
}
|
||||
if len(starts) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make([]zstdFramePart, 0, len(starts))
|
||||
for i, start := range starts {
|
||||
end := len(b)
|
||||
if i+1 < len(starts) {
|
||||
end = starts[i+1]
|
||||
}
|
||||
out = append(out, zstdFramePart{Offset: int64(start), Length: end - start, Data: b[start:end]})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func indexEntryFromMboxRecord(raw []byte, fileOffset, frameLen, innerOffset, innerLen int64) MboxIndexEntry {
|
||||
entry := MboxIndexEntry{FileOffset: fileOffset, FrameLen: frameLen, InnerOffset: innerOffset, InnerLen: innerLen}
|
||||
msg, err := mail.ReadMessage(bytes.NewReader(raw))
|
||||
if err != nil {
|
||||
return entry
|
||||
}
|
||||
entry.MessageID = normalizeMessageID(msg.Header.Get("Message-ID"))
|
||||
if entry.MessageID == "" {
|
||||
entry.MessageID = messageID(raw)
|
||||
}
|
||||
entry.Subject = decodeHeader(msg.Header.Get("Subject"))
|
||||
entry.From = decodeHeader(msg.Header.Get("From"))
|
||||
entry.Date = decodeHeader(msg.Header.Get("Date"))
|
||||
return entry
|
||||
}
|
||||
|
||||
func unescapeMboxMessage(raw []byte) []byte {
|
||||
raw = bytes.ReplaceAll(raw, []byte("\r\n"), []byte("\n"))
|
||||
lines := bytes.Split(raw, []byte("\n"))
|
||||
var b bytes.Buffer
|
||||
for i, line := range lines {
|
||||
if i > 0 {
|
||||
_ = b.WriteByte('\n')
|
||||
}
|
||||
if bytes.HasPrefix(line, []byte(">From ")) {
|
||||
line = line[1:]
|
||||
}
|
||||
_, _ = b.Write(line)
|
||||
}
|
||||
return bytes.TrimRight(b.Bytes(), "\n")
|
||||
}
|
||||
|
||||
func readMboxMessagesBytes(b []byte) [][]byte {
|
||||
b = bytes.ReplaceAll(b, []byte("\r\n"), []byte("\n"))
|
||||
lines := bytes.Split(b, []byte("\n"))
|
||||
|
|
@ -278,11 +512,12 @@ func readMboxMessageFromIndex(path string, index int) ([]byte, bool, error) {
|
|||
if err != nil {
|
||||
return nil, true, err
|
||||
}
|
||||
msgs := readMboxMessagesBytes(record)
|
||||
if len(msgs) == 0 {
|
||||
start := entry.InnerOffset
|
||||
end := entry.InnerOffset + entry.InnerLen
|
||||
if start < 0 || end > int64(len(record)) || start >= end {
|
||||
return nil, true, fmt.Errorf("message index out of range")
|
||||
}
|
||||
return msgs[0], true, nil
|
||||
return unescapeMboxMessage(record[start:end]), true, nil
|
||||
}
|
||||
|
||||
func mboxIndexContext(path string) (int64, string, bool) {
|
||||
|
|
|
|||
|
|
@ -118,3 +118,100 @@ func TestZstdMboxRoundTripAndIndexRead(t *testing.T) {
|
|||
t.Fatalf("unexpected message body: %q", raw)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPlainMboxPartialIndexIsRebuiltBeforeList(t *testing.T) {
|
||||
oldCfg, oldDB := Cfg, DB
|
||||
root := t.TempDir()
|
||||
t.Cleanup(func() {
|
||||
if DB != nil {
|
||||
_ = DB.Close()
|
||||
}
|
||||
Cfg, DB = oldCfg, oldDB
|
||||
})
|
||||
|
||||
Cfg = Config{
|
||||
DBPath: filepath.Join(root, "mail-graveyard.db"),
|
||||
MboxRoot: filepath.Join(root, "backup"),
|
||||
MboxCompression: "none",
|
||||
}
|
||||
if err := ConnectDB(true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := SaveArchiveMailbox("archive"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := SaveAccount(Account{
|
||||
Name: "test-account",
|
||||
SrcHost: "source.example",
|
||||
SrcPort: 993,
|
||||
SrcSecurity: "tls",
|
||||
SrcUser: "source@example.com",
|
||||
SrcPass: "x",
|
||||
DstHost: "target.example",
|
||||
DstPort: 993,
|
||||
DstSecurity: "tls",
|
||||
DstUser: "target@example.com",
|
||||
DstPass: "x",
|
||||
MboxDir: "archive",
|
||||
Active: true,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
account, err := GetAccount("test-account")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
writer, err := NewMboxWriter(filepath.Join(Cfg.MboxRoot, "archive"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
first := testRawMessage("first@example.com", "First")
|
||||
firstInfo, err := writer.Append("INBOX", first)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := SaveMboxIndex(MboxIndexEntry{
|
||||
AccountID: account.ID,
|
||||
Folder: "INBOX",
|
||||
MessageID: first.MessageID,
|
||||
Subject: firstInfo.Subject,
|
||||
From: firstInfo.From,
|
||||
Date: firstInfo.Date,
|
||||
FileOffset: firstInfo.FileOffset,
|
||||
FrameLen: firstInfo.FrameLen,
|
||||
InnerOffset: firstInfo.InnerOffset,
|
||||
InnerLen: firstInfo.InnerLen,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := UpdateMboxIndexState(account.ID, "INBOX", firstInfo.FileOffset+firstInfo.FrameLen); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := writer.Append("INBOX", testRawMessage("second@example.com", "Second")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
entries, err := ReadMboxList(firstInfo.Path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(entries) != 2 {
|
||||
t.Fatalf("expected rebuilt full list with 2 entries, got %#v", entries)
|
||||
}
|
||||
if entries[0].Subject != "First" || entries[1].Subject != "Second" {
|
||||
t.Fatalf("unexpected entries after reindex: %#v", entries)
|
||||
}
|
||||
}
|
||||
|
||||
func testRawMessage(id, subject string) RawMessage {
|
||||
return RawMessage{
|
||||
MessageID: id,
|
||||
InternalDate: time.Date(2026, 7, 14, 10, 11, 12, 0, time.UTC),
|
||||
Body: []byte("Message-ID: <" + id + ">\r\n" +
|
||||
"From: Sender <sender@example.com>\r\n" +
|
||||
"Subject: " + subject + "\r\n" +
|
||||
"Date: Tue, 14 Jul 2026 10:11:12 +0000\r\n" +
|
||||
"\r\n" +
|
||||
"Hello\r\n"),
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -114,12 +114,17 @@ func migrateAccount(a Account, selectedFolders map[string]bool) error {
|
|||
_ = finishJob(jobID, 0, 0, 1, "error")
|
||||
return err
|
||||
}
|
||||
folderMap, err := FolderMap(a.ID)
|
||||
if err != nil {
|
||||
_ = finishJob(jobID, 0, 0, 1, "error")
|
||||
return err
|
||||
}
|
||||
total, done, errs := 0, 0, 0
|
||||
for _, folder := range folders {
|
||||
if len(selectedFolders) > 0 && !selectedFolders[folder.Name] {
|
||||
continue
|
||||
}
|
||||
dstFolder := MapSourceToTarget(folder.Name, folder.Attrs, folder.Delim, dst.Delim(), targets, "")
|
||||
dstFolder := MapSourceToTarget(folder.Name, folder.Attrs, folder.Delim, dst.Delim(), targets, folderMap[folder.Name])
|
||||
if dstFolder == "" {
|
||||
dstFolder = folder.Name
|
||||
}
|
||||
|
|
@ -173,6 +178,12 @@ func migrateAccount(a Account, selectedFolders map[string]bool) error {
|
|||
log.Printf("%s %s mbox index error %s: %v", a.Name, folder.Name, m.MessageID, err)
|
||||
return nil
|
||||
}
|
||||
if err := UpdateMboxIndexState(a.ID, folder.Name, appendInfo.FileOffset+appendInfo.FrameLen); err != nil {
|
||||
errs++
|
||||
folderErrs++
|
||||
log.Printf("%s %s mbox index state error %s: %v", a.Name, folder.Name, m.MessageID, err)
|
||||
return nil
|
||||
}
|
||||
if err := MarkCopied(a.ID, folder.Name, m.MessageID); err != nil {
|
||||
errs++
|
||||
folderErrs++
|
||||
|
|
|
|||
140
backend/13-target-dedup.go
Normal file
140
backend/13-target-dedup.go
Normal file
|
|
@ -0,0 +1,140 @@
|
|||
package backend
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type TargetDedupReport struct {
|
||||
Account string
|
||||
Apply bool
|
||||
Folders int
|
||||
Groups int
|
||||
ExtraCopies int
|
||||
WithoutID int
|
||||
Deleted int
|
||||
FolderReports []TargetDedupFolderReport
|
||||
}
|
||||
|
||||
type TargetDedupFolderReport struct {
|
||||
Folder string
|
||||
Groups int
|
||||
ExtraCopies int
|
||||
WithoutID int
|
||||
Deletes []TargetDedupDelete
|
||||
}
|
||||
|
||||
type TargetDedupDelete struct {
|
||||
UID uint32
|
||||
MessageID string
|
||||
}
|
||||
|
||||
func DedupTarget(name string, folders []string, apply bool) (TargetDedupReport, error) {
|
||||
a, err := GetAccount(name)
|
||||
if err != nil {
|
||||
return TargetDedupReport{}, err
|
||||
}
|
||||
dst, err := OpenIMAPTarget(a)
|
||||
if err != nil {
|
||||
return TargetDedupReport{}, err
|
||||
}
|
||||
defer dst.Close()
|
||||
allFolders, err := dst.Folders()
|
||||
if err != nil {
|
||||
return TargetDedupReport{}, err
|
||||
}
|
||||
selected := selectedFolderSet(folders)
|
||||
report := TargetDedupReport{Account: name, Apply: apply}
|
||||
for _, folder := range allFolders {
|
||||
if len(selected) > 0 && !selected[folder.Name] {
|
||||
continue
|
||||
}
|
||||
folderReport, err := dedupTargetFolder(dst, folder.Name, apply)
|
||||
if err != nil {
|
||||
return report, err
|
||||
}
|
||||
report.Folders++
|
||||
report.Groups += folderReport.Groups
|
||||
report.ExtraCopies += folderReport.ExtraCopies
|
||||
report.WithoutID += folderReport.WithoutID
|
||||
report.Deleted += len(folderReport.Deletes)
|
||||
report.FolderReports = append(report.FolderReports, folderReport)
|
||||
}
|
||||
log.Printf("target dedup %s apply=%v folders=%d groups=%d extra=%d deleted=%d without_id=%d",
|
||||
report.Account, report.Apply, report.Folders, report.Groups, report.ExtraCopies, report.Deleted, report.WithoutID)
|
||||
return report, nil
|
||||
}
|
||||
|
||||
func dedupTargetFolder(dst TargetMailbox, folder string, apply bool) (TargetDedupFolderReport, error) {
|
||||
report := TargetDedupFolderReport{Folder: folder}
|
||||
byID := map[string][]uint32{}
|
||||
for offset := 0; ; offset += mailboxListLimit {
|
||||
headers, err := dst.Headers(folder, mailboxListLimit, offset)
|
||||
if err != nil {
|
||||
return report, fmt.Errorf("%s headers: %w", folder, err)
|
||||
}
|
||||
if len(headers) == 0 {
|
||||
break
|
||||
}
|
||||
for _, header := range headers {
|
||||
id := normalizeMessageID(header.MessageID)
|
||||
if id == "" {
|
||||
report.WithoutID++
|
||||
continue
|
||||
}
|
||||
byID[id] = append(byID[id], header.UID)
|
||||
}
|
||||
}
|
||||
var deleteUIDs []uint32
|
||||
for id, uids := range byID {
|
||||
if len(uids) < 2 {
|
||||
continue
|
||||
}
|
||||
sort.Slice(uids, func(i, j int) bool { return uids[i] < uids[j] })
|
||||
report.Groups++
|
||||
report.ExtraCopies += len(uids) - 1
|
||||
for _, uid := range uids[1:] {
|
||||
report.Deletes = append(report.Deletes, TargetDedupDelete{UID: uid, MessageID: id})
|
||||
deleteUIDs = append(deleteUIDs, uid)
|
||||
}
|
||||
}
|
||||
sort.Slice(report.Deletes, func(i, j int) bool {
|
||||
if report.Deletes[i].MessageID == report.Deletes[j].MessageID {
|
||||
return report.Deletes[i].UID < report.Deletes[j].UID
|
||||
}
|
||||
return strings.Compare(report.Deletes[i].MessageID, report.Deletes[j].MessageID) < 0
|
||||
})
|
||||
if apply && len(deleteUIDs) > 0 {
|
||||
sort.Slice(deleteUIDs, func(i, j int) bool { return deleteUIDs[i] < deleteUIDs[j] })
|
||||
for _, del := range report.Deletes {
|
||||
log.Printf("target dedup delete folder=%s uid=%d message_id=%s", folder, del.UID, del.MessageID)
|
||||
}
|
||||
if err := dst.DeleteUIDs(folder, deleteUIDs); err != nil {
|
||||
return report, fmt.Errorf("%s delete uids: %w", folder, err)
|
||||
}
|
||||
}
|
||||
if report.Groups > 0 || report.WithoutID > 0 {
|
||||
log.Printf("target dedup folder=%s groups=%d extra=%d without_id=%d apply=%v", folder, report.Groups, report.ExtraCopies, report.WithoutID, apply)
|
||||
}
|
||||
return report, nil
|
||||
}
|
||||
|
||||
func LogTargetDedupReport(report TargetDedupReport) {
|
||||
mode := "DRY-RUN"
|
||||
if report.Apply {
|
||||
mode = "APPLY"
|
||||
}
|
||||
log.Printf("target dedup report %s account=%s folders=%d groups=%d extra=%d deleted=%d without_id=%d",
|
||||
mode, report.Account, report.Folders, report.Groups, report.ExtraCopies, report.Deleted, report.WithoutID)
|
||||
for _, folder := range report.FolderReports {
|
||||
if folder.Groups == 0 && folder.WithoutID == 0 {
|
||||
continue
|
||||
}
|
||||
log.Printf("target dedup report folder=%s groups=%d extra=%d without_id=%d", folder.Folder, folder.Groups, folder.ExtraCopies, folder.WithoutID)
|
||||
for _, del := range folder.Deletes {
|
||||
log.Printf("target dedup report candidate folder=%s uid=%d message_id=%s", folder.Folder, del.UID, del.MessageID)
|
||||
}
|
||||
}
|
||||
}
|
||||
165
korrektur-brief.md
Normal file
165
korrektur-brief.md
Normal file
|
|
@ -0,0 +1,165 @@
|
|||
# Codex-Brief — Korrekturen aus dem zstd-Messlauf
|
||||
|
||||
Vier Befunde aus meinem Messlauf (14.07.). Reihenfolge = Priorität.
|
||||
**1 und 2 sind Pflicht vor der ersten echten `@dr-gold.de`-Migration.**
|
||||
|
||||
---
|
||||
|
||||
## 1. 🔴 REGRESSION: Alte plain-Archive sind im Viewer unsichtbar
|
||||
|
||||
**Befund:** Die Nachrichtenliste kommt jetzt aus `mbox_index`. Archive, die
|
||||
**vor** dem zstd-Deploy geschrieben wurden, haben keine Index-Zeilen:
|
||||
|
||||
```
|
||||
mbox_index: codex-abnahme.../INBOX -> 1 Zeile (Datei enthaelt 207 Mails)
|
||||
Viewer zeigt: "INBOX (1)"
|
||||
```
|
||||
|
||||
Das Perfide: Es zeigt **„(1)" statt eines Fehlers**. Ein halb-wahres Ergebnis ist
|
||||
schlimmer als gar keins — es sieht aus, als funktioniere es. Der im zstd-Brief
|
||||
geforderte Fallback auf den alten Parse-Pfad fehlt.
|
||||
|
||||
**Fix — die mbox ist append-only, also reicht ein Fortschritts-Marker pro Datei:**
|
||||
|
||||
```sql
|
||||
CREATE TABLE IF NOT EXISTS mbox_index_state(
|
||||
account_id INTEGER NOT NULL,
|
||||
folder TEXT NOT NULL,
|
||||
indexed_bytes INTEGER NOT NULL DEFAULT 0, -- bis hierhin ist die Datei indiziert
|
||||
UNIQUE(account_id, folder)
|
||||
);
|
||||
```
|
||||
|
||||
- **Nachindizieren:** Wenn `dateigroesse > indexed_bytes` → ab `indexed_bytes`
|
||||
bis EOF scannen, Index-Zeilen anhaengen, `indexed_bytes` fortschreiben.
|
||||
Deckt **leeren** und **teilweisen** Index gleichermassen ab.
|
||||
- Der **Migrationspfad** schreibt `indexed_bytes` beim Append mit fort — dann
|
||||
muss im Normalbetrieb nie nachgescannt werden.
|
||||
- **CLI `--reindex <konto|all>`** fuer den Bestand.
|
||||
- Der Scanner muss **beide** Formate koennen: plain (`From `-Trennzeilen) und
|
||||
`.zst` (Frame fuer Frame entpacken, dann innerhalb des Frames an den
|
||||
`From `-Trennzeilen splitten). Das ist genau die Logik, die Punkt 4 spaeter
|
||||
auch fuer Batch-Frames braucht — einmal richtig bauen.
|
||||
- **Nie wieder still halbe Wahrheiten:** Wenn `dateigroesse != indexed_bytes` und
|
||||
nicht nachindiziert werden kann → **Fehler zeigen**, keine gekuerzte Liste.
|
||||
|
||||
---
|
||||
|
||||
## 2. 🔴 Ziel-Dedup-Werkzeug (die Feuerwehr)
|
||||
|
||||
**Befund:** In `vdevop-03/INBOX` existieren **199 Message-IDs doppelt** — die
|
||||
INBOX wurde irgendwann komplett ein zweites Mal migriert. Niemand hat es gemerkt,
|
||||
weil im Log nur `copied=199` stand und wie normaler Fortschritt aussah.
|
||||
|
||||
Die *Ursachen* sind inzwischen dicht (Message-ID-Normalisierung, DB-Guard,
|
||||
Watch-Fail-Fast). **Aber es gibt kein Werkzeug, das Dubletten im Ziel findet oder
|
||||
repariert.** Wenn das bei einem echten dr-gold-Postfach passiert, gibt es keinen
|
||||
Ausweg — und das merkt man erst, wenn es zu spaet ist.
|
||||
|
||||
**Fix:** `--dedup-target <konto> [--folders ...] [--apply]` (+ Admin-Route).
|
||||
|
||||
- Ordner scannen, **header-only** (`Headers()`-Pfad, keine Bodies!), nach
|
||||
normalisierter Message-ID gruppieren.
|
||||
- Pro Gruppe mit >1 Kopie: **die niedrigste UID behalten** (die aelteste, zuerst
|
||||
eingespielte), den Rest `\Deleted` + `EXPUNGE`.
|
||||
|
||||
**Sicherheitsregeln — das Ding loescht fremde Mail, ein Bug hier ist schlimmer
|
||||
als die Dubletten:**
|
||||
1. **Dry-Run ist Default.** Ohne `--apply` wird nur berichtet: pro Ordner, wie
|
||||
viele Gruppen, wie viele ueberzaehlige Kopien, welche UIDs.
|
||||
2. **Nie die letzte Kopie loeschen.** Pro Message-ID bleibt immer genau eine.
|
||||
3. **Mails ohne Message-ID nie anfassen** — nicht sicher dedupliizierbar.
|
||||
Separat auszaehlen und berichten.
|
||||
4. **Jede Loeschung protokollieren** (Konto, Ordner, UID, Message-ID) — es
|
||||
braucht eine Beweiskette.
|
||||
5. Nur `requireAdmin`.
|
||||
|
||||
---
|
||||
|
||||
## 3. 🟡 `folder_map` ist toter Code
|
||||
|
||||
Die Tabelle wird angelegt, aber **nirgends gelesen**:
|
||||
[`07-migrate.go:122`](backend/07-migrate.go) uebergibt den Override hart als `""`:
|
||||
|
||||
```go
|
||||
dstFolder := MapSourceToTarget(folder.Name, folder.Attrs, folder.Delim, dst.Delim(), targets, "")
|
||||
// ^^ immer leer
|
||||
```
|
||||
|
||||
`MapSourceToTarget` hat den Parameter, `11-folders.go` dokumentiert „manuelles
|
||||
folder_map hat Vorrang" — es tut es nur nicht. Ein dokumentiertes Feature, das
|
||||
stillschweigend nichts tut. (Genau daran ist mein Messlauf haengengeblieben: die
|
||||
Umleitung wurde ignoriert und 207 Mails landeten im echten Ziel-Ordner.)
|
||||
|
||||
**Fix:** `folder_map` pro Konto laden und als `override` durchreichen. Plus
|
||||
Pflege in der UI (Konten-Formular: Quell-Ordner → Ziel-Ordner).
|
||||
|
||||
---
|
||||
|
||||
## 4. 🟢 Kompressionsrate — gemessen, und die Entscheidung dazu
|
||||
|
||||
| | Groesse | Faktor |
|
||||
|---|---|---|
|
||||
| plain mbox (207 Mails) | 4.372.002 | — |
|
||||
| **zstd, Frame pro Mail (heute)** | **1.470.412** | **2,97×** |
|
||||
| zstd, ganze Datei am Stueck | 113.564 | 38,50× |
|
||||
|
||||
Der Abstand ist die **Cross-Message-Redundanz**, die Frame-pro-Mail wegwirft.
|
||||
(Mein Korpus uebertreibt ihn — 199 fast identische Benachrichtigungsmails. Auf
|
||||
echter Mail waere die Luecke kleiner, aber real.)
|
||||
|
||||
**Was wir NICHT machen: zstd-Dictionary.** Das waere technisch der eleganteste
|
||||
Weg (genau dafuer sind Dictionaries da), aber es macht das Archiv **abhaengig von
|
||||
einer externen Datei**: Dictionary weg → Archiv unlesbar. Fuer ein Beweis-Archiv,
|
||||
das **6–10 Jahre** halten und im Streitfall herausgegeben werden muss, ist
|
||||
„`zstd -d` funktioniert einfach, ohne Beiwerk" mehr wert als jeder
|
||||
Kompressionsfaktor. **Selbstgenuegsamkeit schlaegt Rate.**
|
||||
|
||||
**Was wir machen koennen: Batch-Frames mit kleinem Deckel (2 MB).**
|
||||
Der Index (`inner_offset`/`inner_len`) ist bereits dafuer gebaut.
|
||||
|
||||
**Aber die Durability-Regel ist dabei nicht verhandelbar:**
|
||||
Bei Batching ist eine Mail nach `mbox.Append` **noch nicht auf der Platte** — sie
|
||||
liegt im Frame-Puffer. Deshalb:
|
||||
|
||||
```
|
||||
Frame voll (oder Ordner zu Ende)
|
||||
-> Frame komprimieren
|
||||
-> KOMPLETTEN Frame an die Datei anhaengen + fsync
|
||||
-> DANN erst: Index-Zeilen schreiben
|
||||
-> DANN erst: MarkCopied fuer alle Mails des Batches (eine Transaktion)
|
||||
```
|
||||
`MarkCopied` darf **niemals** vor dem fsync passieren — sonst gilt eine Mail als
|
||||
gesichert, die nie im Archiv ankam. Das waere der schlimmstmoegliche Bug fuer ein
|
||||
Beweis-Archiv.
|
||||
|
||||
Nebenwirkung, die man kennen muss: Ein Absturz mitten im Batch laesst bis zu
|
||||
2 MB Mails unmarkiert → sie werden beim naechsten Lauf erneut ins **IMAP-Ziel**
|
||||
kopiert → Dubletten dort. Dasselbe Fenster wie heute, nur groesser (Batch statt
|
||||
Einzelmail). **Deshalb der kleine Deckel (2 MB, nicht 8) — und deshalb ist
|
||||
Punkt 2 (Dedup-Werkzeug) die Voraussetzung dafuer.**
|
||||
|
||||
Priorität: **niedriger als 1–3.** 2,97× ist bereits ein Gewinn, und die Vorschau
|
||||
wurde durch den Index sogar **6× schneller** (1,8 ms statt 10,8 ms). Korrektheit
|
||||
zuerst.
|
||||
|
||||
---
|
||||
|
||||
## Abnahme
|
||||
|
||||
1. `go test ./...` gruen.
|
||||
2. **Reindex:** `--reindex all` → der Viewer zeigt fuer das bestehende
|
||||
plain-Archiv wieder **alle 207** Mails (statt „(1)"). Liste und Oeffnen
|
||||
funktionieren fuer plain **und** `.zst`.
|
||||
3. **Dedup, Dry-Run:** `--dedup-target codex-abnahme-vdevop02-to-vdevop03`
|
||||
meldet fuer `INBOX` **199 Gruppen mit je einer ueberzaehligen Kopie**
|
||||
(das ist der bekannte, echte Bestand — ein guter Selbsttest).
|
||||
Mit `--apply` bleiben danach genau 365 eindeutige Mails uebrig.
|
||||
**Ohne `--apply` darf sich nichts aendern.**
|
||||
4. **folder_map:** Ein Eintrag `INBOX → TEST-ZIEL` fuehrt im Log zu
|
||||
`INBOX -> TEST-ZIEL` (heute: `INBOX -> INBOX`).
|
||||
5. **Round-trip bleibt byte-identisch** — auch nach jeder Aenderung an der
|
||||
Frame-Struktur: `zstd -d x.mbox.zst` == plain-mbox, `cmp`-sauber.
|
||||
Das ist die Linie, die nie fallen darf.
|
||||
|
||||
Die Messungen (Rate, Vorschau-Latenz, Round-trip) fahre ich danach erneut.
|
||||
22
main.go
22
main.go
|
|
@ -23,6 +23,9 @@ func main() {
|
|||
watchInterval := flag.Duration("watch-interval", 60*time.Second, "Pause zwischen Watch-Durchlaeufen")
|
||||
seedAccount := flag.String("seed-account", "", "Konto aus lokaler JSON-Datei in die DB schreiben (nur CLI/Test)")
|
||||
checkAccount := flag.String("check-account", "", "Konto-Logins pruefen und INBOX zaehlen, ohne zu kopieren")
|
||||
reindex := flag.String("reindex", "", "Archiv-mbox-Index fuer Konto/Archiv oder 'all' neu aufbauen")
|
||||
dedupTarget := flag.String("dedup-target", "", "Ziel-Postfach nach Message-ID-Dubletten scannen (Dry-Run ohne --apply)")
|
||||
apply := flag.Bool("apply", false, "Scharfe Ausfuehrung fuer riskante Wartungsbefehle wie --dedup-target")
|
||||
initDB := flag.Bool("init-db", false, "SQLite-DB neu anlegen, falls db_path noch nicht existiert")
|
||||
flag.Parse()
|
||||
|
||||
|
|
@ -44,6 +47,18 @@ func main() {
|
|||
must(backend.CheckAccount(*checkAccount))
|
||||
return
|
||||
}
|
||||
if *reindex != "" {
|
||||
must(backend.ReindexArchives(*reindex))
|
||||
return
|
||||
}
|
||||
if *dedupTarget != "" {
|
||||
report, err := backend.DedupTarget(*dedupTarget, splitOptionalList(*runFolders), *apply)
|
||||
if err != nil {
|
||||
must(err)
|
||||
}
|
||||
backend.LogTargetDedupReport(report)
|
||||
return
|
||||
}
|
||||
|
||||
// CLI-Modus: ohne Web laufen lassen (Cronjob / Cutover).
|
||||
if *runOnce != "" {
|
||||
|
|
@ -78,3 +93,10 @@ func must(err error) {
|
|||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func splitOptionalList(value string) []string {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return nil
|
||||
}
|
||||
return strings.Split(value, ",")
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue