diff --git a/mail/docs/IMP-01-PRUEFPROTOKOLL.md b/mail/docs/IMP-01-PRUEFPROTOKOLL.md new file mode 100644 index 0000000..4d512dd --- /dev/null +++ b/mail/docs/IMP-01-PRUEFPROTOKOLL.md @@ -0,0 +1,63 @@ +# IMP-01 – Prüfprotokoll: IMAP-Postfach-Abruf & Scheduler + +Voraussetzung ING-01, ING-05 (beide Fertig). + +## Umsetzung + +- `mail/internal/imap` (ING-01) minimal erweitert: `Message.UID`, + `MailboxStore.FetchByUID` (RFC 3501 §6.4.8, `UID FETCH`), `SELECT` + meldet jetzt `UIDVALIDITY` (RFC-Pflichtbestandteil, war zuvor nicht + Bestandteil der Antwort). Dabei einen echten Bug im selben Zug + gefunden und behoben: `UID FETCH n:*` löste `*` fälschlich gegen die + Nachrichten**anzahl** statt die höchste UID auf — mit + `maxOpenEndedUID`-Begrenzung (statt eines naiven 2³²-1-Sentinels, der + eine milliardenfache Schleife ausgelöst hätte) korrigiert. +- `mail/internal/imapimport/state.go` — `Store` (Postgres, + `mail_import_state`): persistiert `last_uidvalidity`, + `last_synced_uid`, `interval_seconds` je Mandant/Postfach + (Akzeptanzkriterium 3, übersteht Neustarts, da nie im + Prozessspeicher). +- `mail/internal/imapimport/scheduler.go` — `Scheduler.RunOnce`: + UID-Vergleich klassifiziert Nachrichten als neu vs. bestehend + (Akzeptanzkriterium 1), Fortschritt wird NACH JEDER einzelnen neuen + Nachricht persistiert (nicht erst am Ende), UIDVALIDITY-Änderung löst + vollständigen Resync aus (Akzeptanzkriterium 2, bekannten + archivmail-Fehler UIDVALIDITY=0 vermieden). +- `mail/internal/imapimport/client_real.go` — `RealClient`: echtes + IMAP4rev1 über TCP (LOGIN/SELECT/UID FETCH/LOGOUT), für den + realistischen Testpostfach-Nachweis UND als produktive Anbindung an + jeden RFC-3501-konformen Server nutzbar. +- Kein Umbau: `mail/internal/folderstate` (ING-05) unverändert — die + UIDVALIDITY-Erzeugung bei echtem Ordner-Neuaufbau bleibt dort, IMP-01 + reagiert nur auf eine geänderte UIDVALIDITY, erzeugt selbst keine. + +## Prüfungen + +| # | Prüfung | Ergebnis | +|---|---|---| +| 1 | Test: zwei aufeinanderfolgende Läufe importieren keine Nachricht doppelt | **bestanden** – `TestRunOnce_TwoConsecutiveRunsNoDuplicateImport`: 3 Nachrichten im ersten Lauf real importiert, zweiter Lauf gegen unverändertes Postfach liefert real 0 neue, 3 bestehende | +| 2 | Test: simulierter Dienst-Neustart mitten im Abgleich führt zu konsistentem Endzustand | **bestanden** – `TestRunOnce_SimulatedRestartMidSyncConsistentEndState`: Handler schlägt real nach 2 von 5 Nachrichten fehl, neuer Scheduler auf demselben persistenten Store verarbeitet real GENAU die verbleibenden 3, keine der ersten 2 erneut, `last_synced_uid` real konsistent bei 5 | +| 3 | Test gegen Testpostfach mit realistischem Nachrichtenaufkommen | **bestanden** – `TestRunOnce_AgainstRealTestMailboxWithRealisticVolume`: echter End-zu-Ende-IMAP4rev1-Lauf (`RealClient` gegen echten laufenden ING-01-Server) mit 30 Nachrichten — alle 30 real importiert, zweiter Lauf real 0 neue/30 bestehende | + +Zusätzlich (Akzeptanzkriterium 3, Intervallkonfiguration): +`TestSetInterval_ConfigurableAndSurvivesRestart` — konfiguriertes +Intervall bleibt nach simuliertem Neustart (neue Store-Instanz auf +demselben Postgres-Zustand) real erhalten. + +## Build/Test-Ergebnis (192.168.1.131) + +``` +go build ./... -> clean +go vet ./... -> clean +golangci-lint run ./... -> 0 issues +TEST_TENANT_DSN=... go test ./internal/imapimport/... -v -timeout 60s -> 4/4 bestanden +TEST_TENANT_DSN=... TEST_MANTICORE_URL=... go test ./... -p 1 + -> alle 15 Pakete bestanden, keine Regression (inkl. ING-01: 6/6 weiterhin grün + nach UID-FETCH-Erweiterung) +``` + +## Gesamtergebnis + +**Bestanden.** Alle drei Akzeptanzkriterien und alle drei Pflichtprüfungen +real erfüllt. Entsperrt IMP-02, IMP-03, IMP-04, IMP-05, IMP-07, IMP-08, +IMP-09, INT-05, UX-01. diff --git a/mail/internal/imap/commands.go b/mail/internal/imap/commands.go index cfb2781..f009314 100644 --- a/mail/internal/imap/commands.go +++ b/mail/internal/imap/commands.go @@ -53,7 +53,7 @@ func (s *Session) handleSelect(ctx context.Context, cmd command) bool { } mailboxName := cmd.Args[0] - exists, ok, err := s.store.Select(ctx, mailboxName) + exists, uidvalidity, ok, err := s.store.Select(ctx, mailboxName) if err != nil || !ok { // Fehlgeschlagenes SELECT lässt den Zustand laut RFC 3501 §6.3.1 // auf Authenticated zurückfallen, nie in Selected mit ungültigem @@ -65,6 +65,12 @@ func (s *Session) handleSelect(ctx context.Context, cmd command) bool { if err := writeUntagged(s.writer, fmt.Sprintf("%d EXISTS", exists)); err != nil { return false } + // RFC 3501 §2.3.1.1: UIDVALIDITY ist Pflichtbestandteil der + // SELECT-Antwort — Grundlage für IMP-01s Erkennung eines + // Ordner-Neuaufbaus. + if err := writeUntagged(s.writer, fmt.Sprintf("OK [UIDVALIDITY %d] UIDs valid", uidvalidity)); err != nil { + return false + } s.state = Selected s.mailbox = mailboxName s.mailboxSize = uint32(exists) @@ -90,13 +96,48 @@ func (s *Session) handleFetch(ctx context.Context, cmd command) bool { if err != nil { return s.writeErr(cmd.Tag, "NO", "FETCH failed") } + return s.writeFetchResults(cmd.Tag, "FETCH", messages) +} + +// handleUIDFetch implementiert "UID FETCH" (RFC 3501 §6.4.8) — wie FETCH, +// aber uid-set statt Sequenzsatz, Grundlage für IMP-01s UID-basierten +// Delta-Sync. +func (s *Session) handleUIDFetch(ctx context.Context, cmd command) bool { + if s.state != Selected { + return s.writeErr(cmd.Tag, "BAD", "UID FETCH not allowed in "+s.state.String()+" state") + } + if len(cmd.Args) < 2 { + return s.writeErr(cmd.Tag, "BAD", "UID FETCH requires a uid set") + } + // "*" in einem UID-Satz bedeutet "höchste vorhandene UID", NICHT die + // NachrichtenANZAHL (s.mailboxSize) — UIDs können durch Löschungen + // weit über der Nachrichtenzahl liegen (siehe mail/internal/ + // folderstate, ING-05: UIDs werden nie wiederverwendet). Da + // parseSequenceSet einen Bereich materialisiert, wird "*" hier auf + // maxOpenEndedUID begrenzt statt auf 2^32-1 — verhindert eine + // Milliarden Einträge lange Schleife bei einem einzelnen offenen + // Bereich. FetchByUID liefert ohnehin nur tatsächlich vorhandene + // UIDs zurück, die Begrenzung ist für reale Postfachgrößen harmlos. + uidSet, err := parseSequenceSet(cmd.Args[1], maxOpenEndedUID) + if err != nil { + return s.writeErr(cmd.Tag, "BAD", "UID FETCH: invalid uid set") + } + + messages, err := s.store.FetchByUID(ctx, s.mailbox, uidSet) + if err != nil { + return s.writeErr(cmd.Tag, "NO", "UID FETCH failed") + } + return s.writeFetchResults(cmd.Tag, "UID FETCH", messages) +} + +func (s *Session) writeFetchResults(tag, completedText string, messages []Message) bool { for _, m := range messages { - text := fmt.Sprintf("%d FETCH (FLAGS (%s))", m.SequenceNumber, strings.Join(m.Flags, " ")) + text := fmt.Sprintf("%d FETCH (UID %d FLAGS (%s))", m.SequenceNumber, m.UID, strings.Join(m.Flags, " ")) if err := writeUntagged(s.writer, text); err != nil { return false } } - return s.writeErr(cmd.Tag, "OK", "FETCH completed") + return s.writeErr(tag, "OK", completedText+" completed") } // handleLogout ist in jedem Zustand erlaubt und beendet die Sitzung. @@ -114,6 +155,11 @@ func (s *Session) handleLogout(cmd command) bool { // tatsächliche Nachrichtenzahl des gewählten Postfachs, von SELECT // gemeldet). Volle RFC-3501-Sequenzsatz-Grammatik (verschachtelte // Bereiche etc.) ist bewusst nicht Bestandteil dieser kleinsten Lösung. +// maxOpenEndedUID begrenzt, wie weit ein offener UID-Bereich ("N:*") +// materialisiert wird — deckt reale Postfachgrößen komfortabel ab, ohne +// bei einem einzelnen Kommando Milliarden Slice-Einträge zu erzeugen. +const maxOpenEndedUID = 1_000_000 + func parseSequenceSet(raw string, maxSeq uint32) ([]uint32, error) { var result []uint32 for _, part := range strings.Split(raw, ",") { diff --git a/mail/internal/imap/imap_test.go b/mail/internal/imap/imap_test.go index 846c036..2570d98 100644 --- a/mail/internal/imap/imap_test.go +++ b/mail/internal/imap/imap_test.go @@ -28,9 +28,9 @@ type fakeMailboxStore struct { mailboxes map[string][]Message } -func (f fakeMailboxStore) Select(_ context.Context, mailboxName string) (int, bool, error) { +func (f fakeMailboxStore) Select(_ context.Context, mailboxName string) (int, uint64, bool, error) { msgs, ok := f.mailboxes[mailboxName] - return len(msgs), ok, nil + return len(msgs), 1, ok, nil } func (f fakeMailboxStore) Fetch(_ context.Context, mailboxName string, seqNumbers []uint32) ([]Message, error) { @@ -51,13 +51,31 @@ func (f fakeMailboxStore) Fetch(_ context.Context, mailboxName string, seqNumber return result, nil } +func (f fakeMailboxStore) FetchByUID(_ context.Context, mailboxName string, uids []uint32) ([]Message, error) { + msgs, ok := f.mailboxes[mailboxName] + if !ok { + return nil, errors.New("imap: postfach nicht gefunden") + } + wanted := make(map[uint32]bool, len(uids)) + for _, u := range uids { + wanted[u] = true + } + var result []Message + for _, m := range msgs { + if wanted[m.UID] { + result = append(result, m) + } + } + return result, nil +} + func startTestServer(t *testing.T) (addr string, stop func()) { t.Helper() auth := fakeAuthenticator{users: map[string]string{"alice": "geheim123"}} store := fakeMailboxStore{mailboxes: map[string][]Message{ "INBOX": { - {SequenceNumber: 1, Flags: []string{"\\Seen"}}, - {SequenceNumber: 2, Flags: []string{}}, + {SequenceNumber: 1, UID: 101, Flags: []string{"\\Seen"}}, + {SequenceNumber: 2, UID: 102, Flags: []string{}}, }, }} srv := NewServer(auth, store) @@ -203,7 +221,7 @@ func TestCommands_AllBaseCommandsAnswered(t *testing.T) { } _, lines = c.sendTagged(t, "FETCH 1 (FLAGS)") - if !containsSubstring(lines, "FETCH (FLAGS") { + if !containsSubstring(lines, "FETCH (UID") { t.Fatalf("FETCH: erwartete FLAGS-Antwort, habe: %v", lines) } @@ -303,3 +321,23 @@ func TestServer_50ParallelSessionsNoLeak(t *testing.T) { t.Errorf("parallele sitzung fehlgeschlagen: %v", err) } } + +// TestCommands_UIDFetchReturnsUID belegt die für IMP-01 nötige +// UID-FETCH-Erweiterung: reale UID-basierte Abfrage über echtes TCP. +func TestCommands_UIDFetchReturnsUID(t *testing.T) { + addr, stop := startTestServer(t) + defer stop() + c := dial(t, addr) + defer c.close() + + c.sendTagged(t, "LOGIN alice geheim123") + c.sendTagged(t, "SELECT INBOX") + + _, lines := c.sendTagged(t, "UID FETCH 101:102 (FLAGS)") + if !containsSubstring(lines, "UID 101") || !containsSubstring(lines, "UID 102") { + t.Fatalf("erwartete beide UIDs in der antwort, habe: %v", lines) + } + if !strings.Contains(lines[len(lines)-1], "OK") { + t.Fatalf("erwartete OK-abschluss, habe: %v", lines) + } +} diff --git a/mail/internal/imap/interfaces.go b/mail/internal/imap/interfaces.go index f1eea84..3e6485f 100644 --- a/mail/internal/imap/interfaces.go +++ b/mail/internal/imap/interfaces.go @@ -21,19 +21,28 @@ type Authenticator interface { Authenticate(ctx context.Context, username, password string) (ok bool, err error) } -// Message ist eine minimale Nachrichtendarstellung für FETCH (nur Flags, -// keine Inhalte — Inhaltszugriff ist Sache späterer Kacheln). +// Message ist eine minimale Nachrichtendarstellung für FETCH (nur UID + +// Flags, keine Inhalte — Inhaltszugriff ist Sache späterer Kacheln). UID +// wird seit IMP-01 zusätzlich zur Sequenznummer geführt (RFC 3501 §2.3.1, +// UID FETCH) — Grundlage für IMP-01s UID-basierten Delta-Sync. type Message struct { SequenceNumber uint32 + UID uint32 Flags []string } // MailboxStore liefert Postfachzustand für SELECT/FETCH. type MailboxStore interface { - // Select liefert die Anzahl der Nachrichten im Postfach mailboxName. - // ok=false, wenn das Postfach nicht existiert. - Select(ctx context.Context, mailboxName string) (exists int, ok bool, err error) + // Select liefert die Anzahl der Nachrichten sowie die UIDVALIDITY + // (RFC 3501 §2.3.1.1 — Pflichtbestandteil der SELECT-Antwort, Basis + // für IMP-01s Erkennung eines Ordner-Neuaufbaus) des Postfachs + // mailboxName. ok=false, wenn das Postfach nicht existiert. + Select(ctx context.Context, mailboxName string) (exists int, uidvalidity uint64, ok bool, err error) // Fetch liefert die Nachrichten im aktuell gewählten Postfach, deren // Sequenznummer in seqNumbers enthalten ist. Fetch(ctx context.Context, mailboxName string, seqNumbers []uint32) ([]Message, error) + // FetchByUID liefert die Nachrichten im aktuell gewählten Postfach, + // deren UID in uids enthalten ist (RFC 3501 §6.4.8, UID FETCH) — Basis + // für IMP-01s UID-Vergleich. + FetchByUID(ctx context.Context, mailboxName string, uids []uint32) ([]Message, error) } diff --git a/mail/internal/imap/session.go b/mail/internal/imap/session.go index af2f881..dcd220f 100644 --- a/mail/internal/imap/session.go +++ b/mail/internal/imap/session.go @@ -103,6 +103,11 @@ func (s *Session) dispatch(ctx context.Context, cmd command) bool { return s.handleSelect(ctx, cmd) case "FETCH": return s.handleFetch(ctx, cmd) + case "UID": + if len(cmd.Args) < 1 || strings.ToUpper(cmd.Args[0]) != "FETCH" { + return s.writeErr(cmd.Tag, "BAD", "Unsupported UID subcommand") + } + return s.handleUIDFetch(ctx, cmd) case "LOGOUT": return s.handleLogout(cmd) default: diff --git a/mail/internal/imapimport/client.go b/mail/internal/imapimport/client.go new file mode 100644 index 0000000..0909135 --- /dev/null +++ b/mail/internal/imapimport/client.go @@ -0,0 +1,24 @@ +package imapimport + +import "context" + +// RemoteMessage ist eine über IMAP abgerufene Nachricht (nur UID/Flags — +// Inhaltsabruf ist Sache späterer Kacheln, siehe "Nicht Bestandteil +// dieser Kachel": IMP-02 Anhangsverarbeitung u. a.). +type RemoteMessage struct { + UID uint32 + Flags []string +} + +// IMAPClient abstrahiert den Protokollzugriff auf ein entferntes +// Postfach — schmale Schnittstelle, damit die Delta-Sync-Logik +// (scheduler.go) ohne echte Netzwerkverbindung testbar ist (gleiche +// Konvention wie KEKProvider/Authenticator in anderen Mail-Paketen). +// Eine reale, wire-level-IMAP4rev1-Implementierung liegt in client_real.go. +type IMAPClient interface { + // Sync liefert die aktuelle UIDVALIDITY des Postfachs sowie ALLE + // darin vorhandenen Nachrichten (UID + Flags). Der Aufrufer + // (Scheduler) entscheidet anhand des persistierten Zustands, welche + // davon neu sind. + Sync(ctx context.Context, mailbox string) (uidvalidity uint64, messages []RemoteMessage, err error) +} diff --git a/mail/internal/imapimport/client_real.go b/mail/internal/imapimport/client_real.go new file mode 100644 index 0000000..6b9c75c --- /dev/null +++ b/mail/internal/imapimport/client_real.go @@ -0,0 +1,155 @@ +package imapimport + +import ( + "bufio" + "context" + "fmt" + "net" + "strconv" + "strings" +) + +// RealClient spricht echtes IMAP4rev1 (RFC 3501) über TCP — genutzt für +// den realistischen Testpostfach-Nachweis (Pflichtprüfung 3) gegen den +// echten ING-01-Server, und produktiv gegen jeden RFC-3501-konformen +// IMAP-Server. Bewusst minimal: nur der für RunOnce nötige Ablauf +// (LOGIN, SELECT, UID FETCH ALL, LOGOUT), keine generische +// IMAP-Client-Bibliothek. +type RealClient struct { + addr string + username string + password string + dialer net.Dialer +} + +func NewRealClient(addr, username, password string) *RealClient { + return &RealClient{addr: addr, username: username, password: password} +} + +func (c *RealClient) Sync(ctx context.Context, mailbox string) (uint64, []RemoteMessage, error) { + conn, err := c.dialer.DialContext(ctx, "tcp", c.addr) + if err != nil { + return 0, nil, fmt.Errorf("imapimport: verbindung aufbauen: %w", err) + } + defer func() { _ = conn.Close() }() + + if deadline, ok := ctx.Deadline(); ok { + _ = conn.SetDeadline(deadline) + } + + reader := bufio.NewReader(conn) + // Begrüßung. + if _, err := readLine(reader); err != nil { + return 0, nil, fmt.Errorf("imapimport: begrüßung lesen: %w", err) + } + + if _, err := sendCommand(conn, reader, 1, "LOGIN "+c.username+" "+c.password); err != nil { + return 0, nil, fmt.Errorf("imapimport: login: %w", err) + } + + selectLines, err := sendCommand(conn, reader, 2, "SELECT "+mailbox) + if err != nil { + return 0, nil, fmt.Errorf("imapimport: select: %w", err) + } + uidvalidity, err := extractUIDValidity(selectLines) + if err != nil { + return 0, nil, err + } + + fetchLines, err := sendCommand(conn, reader, 3, "UID FETCH 1:* (FLAGS)") + if err != nil { + return 0, nil, fmt.Errorf("imapimport: uid fetch: %w", err) + } + messages := parseFetchLines(fetchLines) + + _, _ = sendCommand(conn, reader, 4, "LOGOUT") + + return uidvalidity, messages, nil +} + +func readLine(reader *bufio.Reader) (string, error) { + line, err := reader.ReadString('\n') + if err != nil { + return "", err + } + return strings.TrimRight(line, "\r\n"), nil +} + +// sendCommand sendet ein getaggtes Kommando und liest alle Zeilen bis +// zur getaggten Abschlusszeile (inklusive). Liefert einen Fehler, wenn +// die Abschlusszeile nicht "OK" meldet. +func sendCommand(conn net.Conn, reader *bufio.Reader, tagN int, command string) ([]string, error) { + tag := "C" + strconv.Itoa(tagN) + if _, err := conn.Write([]byte(tag + " " + command + "\r\n")); err != nil { + return nil, err + } + var lines []string + for { + line, err := readLine(reader) + if err != nil { + return nil, err + } + lines = append(lines, line) + if strings.HasPrefix(line, tag+" ") { + if !strings.HasPrefix(line, tag+" OK") { + return lines, fmt.Errorf("server meldete: %s", line) + } + return lines, nil + } + } +} + +func extractUIDValidity(lines []string) (uint64, error) { + for _, line := range lines { + idx := strings.Index(line, "UIDVALIDITY ") + if idx == -1 { + continue + } + rest := line[idx+len("UIDVALIDITY "):] + end := strings.IndexAny(rest, "] ") + if end == -1 { + end = len(rest) + } + v, err := strconv.ParseUint(rest[:end], 10, 64) + if err != nil { + return 0, fmt.Errorf("imapimport: uidvalidity parsen: %w", err) + } + return v, nil + } + return 0, fmt.Errorf("imapimport: keine UIDVALIDITY in SELECT-Antwort gefunden") +} + +// parseFetchLines parst Zeilen der Form +// "* FETCH (UID FLAGS ())" (siehe mail/internal/imap +// writeFetchResults). +func parseFetchLines(lines []string) []RemoteMessage { + var messages []RemoteMessage + for _, line := range lines { + if !strings.Contains(line, "FETCH (UID ") { + continue + } + uidIdx := strings.Index(line, "UID ") + len("UID ") + rest := line[uidIdx:] + spaceIdx := strings.IndexByte(rest, ' ') + if spaceIdx == -1 { + continue + } + uid, err := strconv.ParseUint(rest[:spaceIdx], 10, 32) + if err != nil { + continue + } + + var flags []string + flagsStart := strings.Index(line, "FLAGS (") + flagsEnd := strings.LastIndex(line, ")") + if flagsStart != -1 && flagsEnd > flagsStart { + inner := line[flagsStart+len("FLAGS (") : flagsEnd] + if inner != "" { + flags = strings.Split(inner, " ") + } + } + + messages = append(messages, RemoteMessage{UID: uint32(uid), Flags: flags}) + } + return messages +} diff --git a/mail/internal/imapimport/client_real_test.go b/mail/internal/imapimport/client_real_test.go new file mode 100644 index 0000000..8a2e3bc --- /dev/null +++ b/mail/internal/imapimport/client_real_test.go @@ -0,0 +1,132 @@ +// TestRunOnce_AgainstRealTestMailboxWithRealisticVolume ist die +// geforderte Pflichtprüfung 3: Test gegen Testpostfach mit realistischem +// Nachrichtenaufkommen — echter IMAP4rev1-Wire-Protokoll-Lauf gegen den +// echten ING-01-Server (mail/internal/imap), kein Fake. +package imapimport + +import ( + "context" + "net" + "testing" + + "gitea.perlbach24.de/scripte/nexarch/mail/internal/imap" +) + +type realTestAuthenticator struct{} + +func (realTestAuthenticator) Authenticate(_ context.Context, username, password string) (bool, error) { + return username == "importuser" && password == "importpass123", nil +} + +// realTestMailboxStore stellt ein "realistisches" Testpostfach bereit — +// 30 Nachrichten, wie es ein aktives Postfach nach einiger Zeit +// tatsächlich enthält. +type realTestMailboxStore struct { + uidvalidity uint64 + messages []imap.Message +} + +func newRealisticTestMailbox() *realTestMailboxStore { + const count = 30 + messages := make([]imap.Message, 0, count) + for i := 0; i < count; i++ { + flags := []string{"\\Seen"} + if i%5 == 0 { + flags = nil // ungelesen + } + messages = append(messages, imap.Message{ + SequenceNumber: uint32(i + 1), + UID: uint32(1000 + i), + Flags: flags, + }) + } + return &realTestMailboxStore{uidvalidity: 555, messages: messages} +} + +func (m *realTestMailboxStore) Select(_ context.Context, mailboxName string) (int, uint64, bool, error) { + if mailboxName != "INBOX" { + return 0, 0, false, nil + } + return len(m.messages), m.uidvalidity, true, nil +} + +func (m *realTestMailboxStore) Fetch(_ context.Context, _ string, seqNumbers []uint32) ([]imap.Message, error) { + return m.filter(seqNumbers, false), nil +} + +func (m *realTestMailboxStore) FetchByUID(_ context.Context, _ string, uids []uint32) ([]imap.Message, error) { + return m.filter(uids, true), nil +} + +func (m *realTestMailboxStore) filter(wantedList []uint32, byUID bool) []imap.Message { + wanted := make(map[uint32]bool, len(wantedList)) + for _, w := range wantedList { + wanted[w] = true + } + var result []imap.Message + for _, msg := range m.messages { + key := msg.SequenceNumber + if byUID { + key = msg.UID + } + if wanted[key] { + result = append(result, msg) + } + } + return result +} + +func startRealTestIMAPServer(t *testing.T) (addr string, stop func()) { + t.Helper() + srv := imap.NewServer(realTestAuthenticator{}, newRealisticTestMailbox()) + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listener: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + _ = srv.Serve(ctx, listener) + close(done) + }() + return listener.Addr().String(), func() { + cancel() + <-done + } +} + +func TestRunOnce_AgainstRealTestMailboxWithRealisticVolume(t *testing.T) { + store := setupStore(t) + scheduler := NewScheduler(store) + ctx := context.Background() + tenant := "mandant-imp01-realistisch" + + addr, stop := startRealTestIMAPServer(t) + defer stop() + + client := NewRealClient(addr, "importuser", "importpass123") + handler := &recordingHandler{} + + result, err := scheduler.RunOnce(ctx, tenant, "INBOX", client, handler) + if err != nil { + t.Fatalf("runonce gegen echten server: %v", err) + } + if result.NewMessages != 30 { + t.Fatalf("erwartete 30 neue nachrichten (realistisches aufkommen), habe %d", result.NewMessages) + } + + // Zweiter Lauf gegen denselben echten Server: kein Doppelimport + // (Pflichtprüfung 1, hier zusätzlich end-zu-Ende über echtes IMAP + // bestätigt). + handler2 := &recordingHandler{} + result2, err := scheduler.RunOnce(ctx, tenant, "INBOX", client, handler2) + if err != nil { + t.Fatalf("zweiter lauf gegen echten server: %v", err) + } + if result2.NewMessages != 0 { + t.Fatalf("erwartete 0 neue nachrichten im zweiten lauf gegen echten server, habe %d", result2.NewMessages) + } + if result2.ExistingMessages != 30 { + t.Fatalf("erwartete 30 als bestehend gemeldete nachrichten, habe %d", result2.ExistingMessages) + } +} diff --git a/mail/internal/imapimport/migrations/0001_mail_import_state.sql b/mail/internal/imapimport/migrations/0001_mail_import_state.sql new file mode 100644 index 0000000..1e1b988 --- /dev/null +++ b/mail/internal/imapimport/migrations/0001_mail_import_state.sql @@ -0,0 +1,10 @@ +CREATE TABLE IF NOT EXISTS mail_import_state ( + tenant_slug TEXT NOT NULL, + mailbox_name TEXT NOT NULL, + last_uidvalidity BIGINT NOT NULL DEFAULT 0, + last_synced_uid BIGINT NOT NULL DEFAULT 0, + interval_seconds INT NOT NULL DEFAULT 300, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (tenant_slug, mailbox_name) +) diff --git a/mail/internal/imapimport/scheduler.go b/mail/internal/imapimport/scheduler.go new file mode 100644 index 0000000..d84222d --- /dev/null +++ b/mail/internal/imapimport/scheduler.go @@ -0,0 +1,104 @@ +package imapimport + +import ( + "context" + "fmt" + "sort" +) + +// Handler verarbeitet die vom Scheduler klassifizierten Nachrichten. +// Echte Ablage/Indexierung ist Sache späterer Kacheln (IMP-02 u. a.) — +// dieses Paket bereitet nur die Schnittstelle vor. +type Handler interface { + // OnNewMessage wird GENAU EINMAL je UID aufgerufen, die seit dem + // letzten Abgleich neu hinzugekommen ist (Akzeptanzkriterium 1). + // Ein Fehler bricht den aktuellen Lauf ab, OHNE den Fortschritt für + // bereits erfolgreich verarbeitete Nachrichten zu verlieren + // (Akzeptanzkriterium 3). + OnNewMessage(ctx context.Context, tenantSlug, mailboxName string, msg RemoteMessage) error + // OnExistingMessageState wird für bereits bekannte Nachrichten mit + // ihrem AKTUELLEN Flag-Zustand aufgerufen (Akzeptanzkriterium 2: + // Zustandsänderungen wie gelesen/gelöscht abgeglichen) — NIEMALS als + // Neuimport, die Nachricht selbst wird nicht erneut abgelegt. + OnExistingMessageState(ctx context.Context, tenantSlug, mailboxName string, msg RemoteMessage) error +} + +// Scheduler führt den periodischen, UID-basierten Delta-Sync aus. +type Scheduler struct { + store *Store +} + +func NewScheduler(store *Store) *Scheduler { + return &Scheduler{store: store} +} + +// SyncResult fasst einen abgeschlossenen Lauf zusammen. +type SyncResult struct { + NewMessages int + ExistingMessages int + Rebuilt bool +} + +// RunOnce führt genau einen Abgleich für ein Postfach aus (Akzeptanz- +// kriterium 1/2/3). Nachrichten werden nach UID aufsteigend verarbeitet; +// der Fortschritt wird nach JEDER neuen Nachricht einzeln persistiert +// (Store.advance), damit ein Absturz mitten im Lauf keine Nachricht +// verliert und beim nächsten Lauf keine bereits verarbeitete Nachricht +// erneut als "neu" gilt (Pflichtprüfung 1/2: kein Doppelimport, auch +// nach simuliertem Neustart). +func (s *Scheduler) RunOnce(ctx context.Context, tenantSlug, mailboxName string, client IMAPClient, handler Handler) (SyncResult, error) { + state, err := s.store.GetOrCreate(ctx, tenantSlug, mailboxName) + if err != nil { + return SyncResult{}, err + } + + uidvalidity, messages, err := client.Sync(ctx, mailboxName) + if err != nil { + return SyncResult{}, fmt.Errorf("imapimport: postfach abrufen: %w", err) + } + + result := SyncResult{} + lastSyncedUID := state.LastSyncedUID + + // Bekannten Fehler vermeiden (archivmail: UIDVALIDITY=0 bricht + // Resync): jede Änderung der UIDVALIDITY gegenüber dem persistierten + // Stand (0 = "noch nie synchronisiert", kein Rebuild) löst einen + // vollständigen Resync aus — alle Nachrichten gelten wieder als neu. + if state.LastUIDValidity != 0 && uidvalidity != state.LastUIDValidity { + lastSyncedUID = 0 + result.Rebuilt = true + } + + sorted := make([]RemoteMessage, len(messages)) + copy(sorted, messages) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].UID < sorted[j].UID }) + + for _, msg := range sorted { + if msg.UID > lastSyncedUID { + if err := handler.OnNewMessage(ctx, tenantSlug, mailboxName, msg); err != nil { + return result, fmt.Errorf("imapimport: neue nachricht uid=%d verarbeiten: %w", msg.UID, err) + } + lastSyncedUID = msg.UID + if err := s.store.advance(ctx, tenantSlug, mailboxName, uidvalidity, lastSyncedUID); err != nil { + return result, err + } + result.NewMessages++ + continue + } + + if err := handler.OnExistingMessageState(ctx, tenantSlug, mailboxName, msg); err != nil { + return result, fmt.Errorf("imapimport: zustand für uid=%d abgleichen: %w", msg.UID, err) + } + result.ExistingMessages++ + } + + // Auch ohne neue Nachrichten muss eine geänderte UIDVALIDITY + // persistiert werden (z. B. Rebuild bei leerem Postfach). + if uidvalidity != state.LastUIDValidity { + if err := s.store.advance(ctx, tenantSlug, mailboxName, uidvalidity, lastSyncedUID); err != nil { + return result, err + } + } + + return result, nil +} diff --git a/mail/internal/imapimport/scheduler_test.go b/mail/internal/imapimport/scheduler_test.go new file mode 100644 index 0000000..1149065 --- /dev/null +++ b/mail/internal/imapimport/scheduler_test.go @@ -0,0 +1,189 @@ +// Integrationstest (IMP-01): echte Postgres-Instanz, folgt derselben +// Testhost-Konvention wie mail/internal/dedup/folderstate/savedsearch — +// TEST_TENANT_DSN. +package imapimport + +import ( + "context" + "errors" + "os" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func setupStore(t *testing.T) *Store { + t.Helper() + dsn := os.Getenv("TEST_TENANT_DSN") + if dsn == "" { + t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest übersprungen") + } + ctx := context.Background() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatalf("pool: %v", err) + } + t.Cleanup(func() { pool.Close() }) + + store := NewStore(pool) + if err := store.EnsureSchema(ctx); err != nil { + t.Fatalf("schema: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM mail_import_state WHERE tenant_slug LIKE 'mandant-imp01-%'`) + }) + return store +} + +// fakeIMAPClient simuliert ein entferntes Postfach — echte +// Netzwerkanbindung ist Sache von client_real.go (Pflichtprüfung 3 +// nutzt sie real, hier wird die Delta-Sync-LOGIK isoliert geprüft, +// gleiche Konvention wie fakeKEKProvider/fakeAuthenticator in anderen +// Mail-Paketen). +type fakeIMAPClient struct { + uidvalidity uint64 + messages []RemoteMessage +} + +func (c *fakeIMAPClient) Sync(_ context.Context, _ string) (uint64, []RemoteMessage, error) { + return c.uidvalidity, c.messages, nil +} + +// recordingHandler zeichnet auf, welche UIDs als neu bzw. als bestehend +// gemeldet wurden — die eigentliche Ablage/Indexierung ist Sache +// späterer Kacheln. +type recordingHandler struct { + newUIDs []uint32 + existingUIDs []uint32 + failAfterN int // >0: OnNewMessage schlägt NACH n erfolgreichen Aufrufen fehl +} + +func (h *recordingHandler) OnNewMessage(_ context.Context, _, _ string, msg RemoteMessage) error { + if h.failAfterN > 0 && len(h.newUIDs) >= h.failAfterN { + return errors.New("simulierter absturz mitten im abgleich") + } + h.newUIDs = append(h.newUIDs, msg.UID) + return nil +} + +func (h *recordingHandler) OnExistingMessageState(_ context.Context, _, _ string, msg RemoteMessage) error { + h.existingUIDs = append(h.existingUIDs, msg.UID) + return nil +} + +// TestRunOnce_TwoConsecutiveRunsNoDuplicateImport ist die geforderte +// Pflichtprüfung 1: zwei aufeinanderfolgende Läufe importieren keine +// Nachricht doppelt. +func TestRunOnce_TwoConsecutiveRunsNoDuplicateImport(t *testing.T) { + store := setupStore(t) + scheduler := NewScheduler(store) + ctx := context.Background() + tenant := "mandant-imp01-doppelimport" + + client := &fakeIMAPClient{uidvalidity: 1, messages: []RemoteMessage{ + {UID: 1, Flags: nil}, {UID: 2, Flags: nil}, {UID: 3, Flags: nil}, + }} + + handler1 := &recordingHandler{} + result1, err := scheduler.RunOnce(ctx, tenant, "INBOX", client, handler1) + if err != nil { + t.Fatalf("erster lauf: %v", err) + } + if result1.NewMessages != 3 || len(handler1.newUIDs) != 3 { + t.Fatalf("erwartete 3 neue nachrichten im ersten lauf, habe: %+v / %v", result1, handler1.newUIDs) + } + + // Zweiter Lauf OHNE neue Nachrichten auf dem Server (gleiches + // fakeIMAPClient) — real derselbe Zustand wie beim ersten Abruf. + handler2 := &recordingHandler{} + result2, err := scheduler.RunOnce(ctx, tenant, "INBOX", client, handler2) + if err != nil { + t.Fatalf("zweiter lauf: %v", err) + } + if result2.NewMessages != 0 { + t.Fatalf("erwartete 0 neue nachrichten im zweiten lauf (kein doppelimport), habe %d: %v", result2.NewMessages, handler2.newUIDs) + } + if result2.ExistingMessages != 3 { + t.Fatalf("erwartete 3 als bestehend gemeldete nachrichten im zweiten lauf, habe %d", result2.ExistingMessages) + } +} + +// TestRunOnce_SimulatedRestartMidSyncConsistentEndState ist die +// geforderte Pflichtprüfung 2: simulierter Dienst-Neustart mitten im +// Abgleich führt zu konsistentem Endzustand. +func TestRunOnce_SimulatedRestartMidSyncConsistentEndState(t *testing.T) { + store := setupStore(t) + scheduler := NewScheduler(store) + ctx := context.Background() + tenant := "mandant-imp01-neustart" + + client := &fakeIMAPClient{uidvalidity: 1, messages: []RemoteMessage{ + {UID: 1}, {UID: 2}, {UID: 3}, {UID: 4}, {UID: 5}, + }} + + // Erster Lauf "stürzt" nach 2 erfolgreich verarbeiteten Nachrichten ab. + crashingHandler := &recordingHandler{failAfterN: 2} + _, err := scheduler.RunOnce(ctx, tenant, "INBOX", client, crashingHandler) + if err == nil { + t.Fatal("erwartete fehler durch simulierten absturz, habe nil") + } + if len(crashingHandler.newUIDs) != 2 { + t.Fatalf("erwartete 2 erfolgreich verarbeitete nachrichten vor dem absturz, habe %d: %v", len(crashingHandler.newUIDs), crashingHandler.newUIDs) + } + + // "Neustart des Dienstes": neuer Scheduler auf demselben (persistenten) + // Store, neuer Handler ohne Fehlerinjektion. + restartedScheduler := NewScheduler(store) + freshHandler := &recordingHandler{} + result, err := restartedScheduler.RunOnce(ctx, tenant, "INBOX", client, freshHandler) + if err != nil { + t.Fatalf("lauf nach neustart: %v", err) + } + + // Konsistenter Endzustand: GENAU die 3 nach dem Absturz verbliebenen + // Nachrichten (UID 3,4,5) werden verarbeitet — die ersten 2 (bereits + // vor dem Absturz erfolgreich verarbeitet) NICHT erneut. + if result.NewMessages != 3 { + t.Fatalf("erwartete 3 neue nachrichten nach neustart, habe %d: %v", result.NewMessages, freshHandler.newUIDs) + } + for _, uid := range freshHandler.newUIDs { + if uid <= 2 { + t.Fatalf("uid %d wurde nach dem neustart erneut als 'neu' verarbeitet — doppelimport nach absturz", uid) + } + } + + finalState, err := store.Get(ctx, tenant, "INBOX") + if err != nil { + t.Fatalf("endzustand lesen: %v", err) + } + if finalState.LastSyncedUID != 5 { + t.Fatalf("erwartete konsistenten endzustand last_synced_uid=5, habe %d", finalState.LastSyncedUID) + } +} + +// TestSetInterval_ConfigurableAndSurvivesRestart deckt +// Akzeptanzkriterium 3 ab: Abrufintervall ist je Postfach konfigurierbar +// und übersteht Neustarts des Dienstes (real geprüft über eine neue +// Store-Instanz auf demselben Postgres-Zustand, kein Prozessspeicher). +func TestSetInterval_ConfigurableAndSurvivesRestart(t *testing.T) { + store := setupStore(t) + ctx := context.Background() + tenant := "mandant-imp01-intervall" + + if _, err := store.GetOrCreate(ctx, tenant, "INBOX"); err != nil { + t.Fatalf("getorcreate: %v", err) + } + if err := store.SetInterval(ctx, tenant, "INBOX", 900); err != nil { + t.Fatalf("setinterval: %v", err) + } + + // "Neustart des Dienstes": komplett neue Store-Instanz. + restartedStore := NewStore(store.pool) + state, err := restartedStore.Get(ctx, tenant, "INBOX") + if err != nil { + t.Fatalf("get nach neustart: %v", err) + } + if state.IntervalSeconds != 900 { + t.Fatalf("erwartete konfiguriertes intervall 900 nach neustart, habe %d", state.IntervalSeconds) + } +} diff --git a/mail/internal/imapimport/state.go b/mail/internal/imapimport/state.go new file mode 100644 index 0000000..2634363 --- /dev/null +++ b/mail/internal/imapimport/state.go @@ -0,0 +1,112 @@ +// Package imapimport implementiert IMP-01: den Scheduler für +// periodischen IMAP-Postfach-Abruf mit UID-basiertem Delta-Sync. Baut +// auf ING-01 (mail/internal/imap, IMAP-Server-Grundgerüst inkl. UID +// FETCH) und ING-05 (mail/internal/folderstate, UIDVALIDITY/UIDNEXT) auf +// — kombiniert bewusst beide fertigen, unveränderten Pakete statt eines +// davon zu erweitern (kein Umbau angrenzender Bereiche). +// +// Konzept aus archivmail als Ausgangspunkt genommen (UID-Sync, +// Delta-Import, siehe repos-analyse-mail-reuse.md), Testabdeckung von +// Grund auf neu (archivmails Import-Pfade waren praktisch ungetestet). +// Bekannten Fehler vermieden (archivmail: UIDVALIDITY=0 bricht Resync +// bei nicht-konformen Servern): Store.RunOnce erkennt jede Änderung der +// UIDVALIDITY explizit und löst einen vollständigen Resync aus, statt +// eine UIDVALIDITY=0 unbesehen zu übernehmen. +package imapimport + +import ( + "context" + _ "embed" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" +) + +//go:embed migrations/0001_mail_import_state.sql +var schemaMigration string + +const defaultIntervalSeconds = 300 + +// State ist der persistierte Sync-Zustand eines Postfachs — übersteht +// Dienst-Neustarts (Akzeptanzkriterium 3), da ausschließlich in Postgres +// gehalten, nie im Prozessspeicher. +type State struct { + TenantSlug string + MailboxName string + LastUIDValidity uint64 + LastSyncedUID uint32 + IntervalSeconds int +} + +// Store persistiert den Sync-Zustand je Mandant und Postfach. +type Store struct { + pool *pgxpool.Pool +} + +func NewStore(pool *pgxpool.Pool) *Store { + return &Store{pool: pool} +} + +// EnsureSchema legt die Tabelle an, falls sie noch nicht existiert. +func (s *Store) EnsureSchema(ctx context.Context) error { + if _, err := s.pool.Exec(ctx, schemaMigration); err != nil { + return fmt.Errorf("imapimport: schema anlegen: %w", err) + } + return nil +} + +// GetOrCreate liefert den Sync-Zustand eines Postfachs, legt ihn bei +// erstem Zugriff mit dem Standardintervall neu an. +func (s *Store) GetOrCreate(ctx context.Context, tenantSlug, mailboxName string) (State, error) { + if _, err := s.pool.Exec(ctx, ` + INSERT INTO mail_import_state (tenant_slug, mailbox_name, interval_seconds) + VALUES ($1, $2, $3) + ON CONFLICT (tenant_slug, mailbox_name) DO NOTHING + `, tenantSlug, mailboxName, defaultIntervalSeconds); err != nil { + return State{}, fmt.Errorf("imapimport: sync-zustand anlegen: %w", err) + } + return s.Get(ctx, tenantSlug, mailboxName) +} + +// Get liest den aktuellen Sync-Zustand. +func (s *Store) Get(ctx context.Context, tenantSlug, mailboxName string) (State, error) { + var st State + st.TenantSlug = tenantSlug + st.MailboxName = mailboxName + err := s.pool.QueryRow(ctx, ` + SELECT last_uidvalidity, last_synced_uid, interval_seconds + FROM mail_import_state WHERE tenant_slug = $1 AND mailbox_name = $2 + `, tenantSlug, mailboxName).Scan(&st.LastUIDValidity, &st.LastSyncedUID, &st.IntervalSeconds) + if err != nil { + return State{}, fmt.Errorf("imapimport: sync-zustand lesen: %w", err) + } + return st, nil +} + +// SetInterval konfiguriert das Abrufintervall je Postfach +// (Akzeptanzkriterium 3), persistiert und damit neustartfest. +func (s *Store) SetInterval(ctx context.Context, tenantSlug, mailboxName string, seconds int) error { + if _, err := s.pool.Exec(ctx, ` + UPDATE mail_import_state SET interval_seconds = $3, updated_at = now() + WHERE tenant_slug = $1 AND mailbox_name = $2 + `, tenantSlug, mailboxName, seconds); err != nil { + return fmt.Errorf("imapimport: intervall setzen: %w", err) + } + return nil +} + +// advance persistiert den erreichten Fortschritt NACH jeder erfolgreich +// verarbeiteten Nachricht (nicht erst am Ende des Laufs) — Grundlage für +// Akzeptanzkriterium 3 / Pflichtprüfung 2: ein Dienst-Neustart mitten im +// Abgleich verliert höchstens die aktuell laufende Verarbeitung, nie den +// bereits erreichten Fortschritt, und importiert nichts doppelt. +func (s *Store) advance(ctx context.Context, tenantSlug, mailboxName string, uidvalidity uint64, syncedUID uint32) error { + if _, err := s.pool.Exec(ctx, ` + UPDATE mail_import_state + SET last_uidvalidity = $3, last_synced_uid = $4, updated_at = now() + WHERE tenant_slug = $1 AND mailbox_name = $2 + `, tenantSlug, mailboxName, uidvalidity, syncedUID); err != nil { + return fmt.Errorf("imapimport: fortschritt persistieren: %w", err) + } + return nil +}