diff --git a/docs/MAILBOX-CORE.md b/docs/MAILBOX-CORE.md new file mode 100644 index 0000000..85acbe9 --- /dev/null +++ b/docs/MAILBOX-CORE.md @@ -0,0 +1,57 @@ +# Mailbox core + +The webmail, IMAP, POP3 and SMTP are four ways in to one mailbox. They share +the rules and the data code; none of them has its own copy. + +## Where each rule lives + +| Rule | One place | +|------|-----------| +| Flags, expunge, move, copy, folder delete, counters, quota | `MailStore` (`services/protocols/internal/infrastructure/postgres/mail_store.go`) | +| Folder and message reads | `mail_store_read.go` (same store) | +| Credentials (password, app password, scopes) | `usecase/credentials.go` | +| Sending, and the copy filed in Sent | `ProcessOutboundEmail.Submit` (`FileSentCopy`) | +| Default folders of a mailbox | SQL function `provision_default_folders` (migration 0014) | + +The IMAP repositories, the webmail repository and the TS auth service are thin +adapters: they translate their protocol and call the code above. + +## Invariants + +After any operation, through any port: + +- `folders.total_count` and `unread_count` equal the live messages and the ones + without `\Seen`. +- `folders.highest_modseq` rises only when a message really changed. +- `mailboxes.used_bytes` equals the size of the live messages. +- Deleting twice, or from two ports at once, counts once. + +`mail_state_contract_test.go` checks all of this after every operation through +the IMAP path and the webmail path, and under concurrency. +`e2e/one_mailbox_everywhere_e2e_test.go` does the same over real SMTP, IMAP and +HTTP on one mailbox. + +## Lock order + +Every operation locks **folders first, lowest id first, then the message** +(`lockFolders`). Two moves in opposite directions, or a move racing a flag +change, used to deadlock (SQLSTATE 40P01) because they locked in different +orders. `TestContract_OppositeMovesAndFlagChangesNeverDeadlock` covers it. A new +operation must follow the same order. + +## Adding an operation + +1. Put it in `MailStore`, in one transaction, with `lockFolders`. +2. Make the adapters call it. +3. Add it to the contract test and run it through both paths. + +## Migration 0014 + +Creates `provision_default_folders`, creates the missing default folders of +existing mailboxes, and recomputes folder counters and `used_bytes` from the +messages. It is idempotent and runs with the other migrations at deploy. + +## Not done + +IMAP `APPEND` is still a stub. When it is implemented, drop a client-appended +Sent copy that matches the server's one by `Message-ID`. diff --git a/migrations/0014_mailbox_state_single_source.down.sql b/migrations/0014_mailbox_state_single_source.down.sql new file mode 100644 index 0000000..9dfb621 --- /dev/null +++ b/migrations/0014_mailbox_state_single_source.down.sql @@ -0,0 +1,4 @@ +-- The reconciliation corrected cached numbers and cannot be undone (and has no +-- reason to be); only the function is removed. Folders it added stay: they are +-- ordinary folders by now and may hold mail. +DROP FUNCTION IF EXISTS provision_default_folders(UUID); diff --git a/migrations/0014_mailbox_state_single_source.up.sql b/migrations/0014_mailbox_state_single_source.up.sql new file mode 100644 index 0000000..98d52d4 --- /dev/null +++ b/migrations/0014_mailbox_state_single_source.up.sql @@ -0,0 +1,77 @@ +-- One definition of "a mailbox's standard folders", and a reconciliation of the +-- cached numbers that two code paths used to keep on their own. +-- +-- 1. provision_default_folders +-- +-- The standard folders were created in two places that had already diverged: +-- the console's createMailbox added Reports, the CLI that creates the first +-- administrator did not, so that account had no folder for DMARC and TLS-RPT +-- reports and they fell into its INBOX. Both now call this function, so the +-- list lives here and nowhere else. It is idempotent: it adds only what is +-- missing, matching by name without regard to case. +CREATE OR REPLACE FUNCTION provision_default_folders(p_mailbox_id UUID) +RETURNS VOID +LANGUAGE sql +AS $$ + INSERT INTO folders (mailbox_id, name, special_use) + SELECT p_mailbox_id, f.name, f.special_use + FROM (VALUES ('INBOX', 'inbox'), + ('Sent', 'sent'), + ('Drafts', 'drafts'), + ('Trash', 'trash'), + ('Junk', 'junk'), + ('Archive', 'archive'), + -- Where delivered DMARC and TLS-RPT reports are filed. IMAP has no + -- special-use role for these, so it is matched by name. + ('Reports', NULL)) AS f(name, special_use) + WHERE NOT EXISTS ( + SELECT 1 FROM folders x + WHERE x.mailbox_id = p_mailbox_id + AND x.parent_id IS NULL + AND LOWER(x.name) = LOWER(f.name)); +$$; + +-- Mailboxes created before the function that lack a standard folder get it now. +SELECT provision_default_folders(id) FROM mailboxes; + +-- 2. Reconcile the cached numbers with the rows. +-- +-- Folder counters, and the mailbox's used bytes, are denormalised caches. IMAP +-- and the webmail each maintained them and each got something wrong (IMAP never +-- moved the unread counter; the webmail never gave the quota back), so existing +-- mailboxes carry the drift. Every path now goes through one MailStore that +-- keeps them true; this corrects what is already stored, once. It only touches +-- rows that are actually wrong, so it is safe to run again. +UPDATE folders f + SET unread_count = c.unread, + total_count = c.total + FROM ( + SELECT f2.id, + COUNT(e.id) FILTER ( + WHERE NOT EXISTS ( + SELECT 1 FROM message_flags mf + WHERE mf.message_id = e.id + AND mf.received_at = e.received_at + AND mf.flag = '\Seen') + )::int AS unread, + COUNT(e.id)::int AS total + FROM folders f2 + LEFT JOIN email_messages e + ON e.folder_id = f2.id AND e.expunged_at IS NULL + GROUP BY f2.id + ) AS c + WHERE f.id = c.id + AND (f.unread_count IS DISTINCT FROM c.unread + OR f.total_count IS DISTINCT FROM c.total); + +UPDATE mailboxes m + SET used_bytes = COALESCE(s.total, 0) + FROM mailboxes m2 + LEFT JOIN ( + SELECT mailbox_id, SUM(size_bytes)::bigint AS total + FROM email_messages + WHERE expunged_at IS NULL + GROUP BY mailbox_id + ) AS s ON s.mailbox_id = m2.id + WHERE m.id = m2.id + AND m.used_bytes IS DISTINCT FROM COALESCE(s.total, 0); diff --git a/services/auth/src/cli.test.ts b/services/auth/src/cli.test.ts new file mode 100644 index 0000000..f538aa6 --- /dev/null +++ b/services/auth/src/cli.test.ts @@ -0,0 +1,46 @@ +import { describe, it, expect, afterAll } from "vitest"; + +process.env.LAMBDAMAIL_MASTER_KEY ||= "test-master-key-at-least-16-chars"; +process.env.DATABASE_URL = process.env.TEST_DATABASE_URL ?? ""; + +const hasDatabase = Boolean(process.env.TEST_DATABASE_URL); +const describeDb = hasDatabase ? describe : describe.skip; + +const { createAdmin, CliError } = await import("./cli.js"); +const { query, closePool } = await import("./db.js"); + +const DOMAIN = `cli-${Date.now()}.test`; +const EMAIL = `root@${DOMAIN}`; + +describeDb("create-admin (command line)", () => { + afterAll(async () => { + await query(`DELETE FROM domains WHERE name = $1`, [DOMAIN]); + await closePool(); + }); + + it("gives the first administrator every standard folder, Reports included", async () => { + const log = console.log; + console.log = () => undefined; + try { + await createAdmin(EMAIL, "a-long-enough-password", true); + } finally { + console.log = log; + } + + // This account used to come out without Reports, so its DMARC and TLS-RPT + // reports fell into the INBOX: the console and this command each had their + // own list of folders and the lists had drifted. + const folders = await query<{ name: string; special_use: string | null }>( + `SELECT f.name, f.special_use FROM folders f JOIN mailboxes m ON m.id = f.mailbox_id WHERE m.email_address = $1`, + [EMAIL], + ); + expect(folders.map((f) => f.name).sort()).toEqual(["Archive", "Drafts", "INBOX", "Junk", "Reports", "Sent", "Trash"]); + expect(folders.find((f) => f.name === "Reports")?.special_use).toBeNull(); + expect(folders.find((f) => f.name === "Sent")?.special_use).toBe("sent"); + }); + + it("refuses with a readable error instead of exiting the process", async () => { + await expect(createAdmin(EMAIL, "short", true)).rejects.toBeInstanceOf(CliError); + await expect(createAdmin(EMAIL, "a-long-enough-password", true)).rejects.toThrow(/already exists/); + }); +}); diff --git a/services/auth/src/cli.ts b/services/auth/src/cli.ts index 57fce04..7f3351c 100644 --- a/services/auth/src/cli.ts +++ b/services/auth/src/cli.ts @@ -13,22 +13,19 @@ * node dist/cli.js reset-mfa */ +import { pathToFileURL } from "node:url"; import { query, queryOne, closePool } from "./db.js"; import { hashPassword } from "./crypto.js"; const SYSTEM_ALIASES = ["postmaster", "abuse", "dmarc", "tlsrpt"]; -const STANDARD_FOLDERS: Array<[string, string]> = [ - ["INBOX", "inbox"], - ["Sent", "sent"], - ["Drafts", "drafts"], - ["Trash", "trash"], - ["Junk", "junk"], - ["Archive", "archive"], -]; +/** A refusal the operator should read, as opposed to a bug. */ +export class CliError extends Error {} + +// Thrown rather than exiting here, so the commands can be exercised from a test; +// main() prints it as "error: ..." and exits 1, exactly as this used to. function fail(message: string): never { - console.error(`error: ${message}`); - process.exit(1); + throw new CliError(message); } function parseAddress(email: string): { localPart: string; domain: string } { @@ -37,7 +34,7 @@ function parseAddress(email: string): { localPart: string; domain: string } { return { localPart: email.slice(0, at).toLowerCase(), domain: email.slice(at + 1).toLowerCase() }; } -async function createAdmin(email: string, password: string, force: boolean): Promise { +export async function createAdmin(email: string, password: string, force: boolean): Promise { if (password.length < 12) fail("the password must be at least 12 characters"); const { localPart, domain } = parseAddress(email); @@ -69,13 +66,10 @@ async function createAdmin(email: string, password: string, force: boolean): Pro ); // Delivery needs these: a mailbox with no INBOX cannot receive at all, and - // one with no Junk used to fail the spam path outright. - for (const [name, specialUse] of STANDARD_FOLDERS) { - await query( - `INSERT INTO folders (mailbox_id, name, special_use) VALUES ($1, $2, $3)`, - [mailbox!.id, name, specialUse], - ); - } + // one with no Junk used to fail the spam path outright. The list lives in the + // database (migration 0014) so the console and this command cannot disagree + // about it, as they once did over the Reports folder. + await query(`SELECT provision_default_folders($1)`, [mailbox!.id]); // PLAN.md section 7.4b: the published DNS records name these, and reports // bounce if they do not resolve. @@ -174,10 +168,13 @@ async function main(): Promise { } } -main() - .then(() => closePool()) - .catch(async (err) => { - console.error(`error: ${err instanceof Error ? err.message : String(err)}`); - await closePool().catch(() => undefined); - process.exit(1); - }); +// Only when run as a command, not when a test imports the functions above +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + main() + .then(() => closePool()) + .catch(async (err) => { + console.error(`error: ${err instanceof Error ? err.message : String(err)}`); + await closePool().catch(() => undefined); + process.exit(1); + }); +} diff --git a/services/auth/src/repository.ts b/services/auth/src/repository.ts index 45457a2..0560da4 100644 --- a/services/auth/src/repository.ts +++ b/services/auth/src/repository.ts @@ -1004,18 +1004,9 @@ export async function createMailbox(scope: AdminScope, input: CreateMailboxInput [input.domainId, localPart, email, hash, input.role, input.quotaBytes], ); - await query( - `INSERT INTO folders (mailbox_id, name, special_use) - SELECT $1, f.name, f.special_use - FROM (VALUES ('INBOX','inbox'), ('Sent','sent'), ('Drafts','drafts'), - ('Trash','trash'), ('Junk','junk'), ('Archive','archive'), - -- Where delivered DMARC and TLS-RPT reports are filed. - -- IMAP has no special-use role for these, so it is matched - -- by name. Without the folder the delivery path falls back - -- to INBOX, which is the noise this is here to remove. - ('Reports', NULL)) AS f(name, special_use)`, - [created!.id], - ); + // The standard folders come from one definition in the database (migration + // 0014), shared with the command that creates the first administrator. + await query(`SELECT provision_default_folders($1)`, [created!.id]); return { status: "OK", id: created!.id, email }; } diff --git a/services/protocols/internal/application/usecase/sent_copy_test.go b/services/protocols/internal/application/usecase/sent_copy_test.go index 019783d..52c5146 100644 --- a/services/protocols/internal/application/usecase/sent_copy_test.go +++ b/services/protocols/internal/application/usecase/sent_copy_test.go @@ -137,3 +137,51 @@ func TestSubmit_WorksWithoutASentCopyRepository(t *testing.T) { t.Fatalf("Submit without a Sent repository: %v", err) } } + +// One rule for every path: a message sent from the webmail and the same message +// sent by an SMTP client leave the same thing behind in Sent. They used to differ +// completely - the webmail filed a copy, SMTP filed nothing - because the rule +// lived in the webmail alone. +func TestSentCopyIsTheSameWhicheverPathSent(t *testing.T) { + mailboxID := uuid.New() + messages := &recordingMessageRepository{} + submission := NewProcessOutboundEmailUseCase(nil, &countingOutboundRepo{}, &capturingBlobStorage{}, nil) + submission.SetSentCopies(messages) + + account := &port.MailboxAuth{ID: mailboxID, EmailAddress: "me@example.test", DomainName: "example.test", MaxRecipientsPerHour: 100} + web := &WebmailUseCase{ + repo: &stubWebmailRepo{mailboxID: mailboxID.String()}, + auth: &stubAuthRepo{account: account}, + submission: submission, + localHost: "mail.example.test", + } + + if err := web.Send(context.Background(), ComposeInput{From: "me@example.test", To: []string{"you@remote.test"}, Subject: "Aviso", Text: "hi"}); err != nil { + t.Fatalf("webmail send: %v", err) + } + + smtp := ProcessOutboundEmailInput{ + MailboxID: mailboxID, SenderAddr: "me@example.test", DomainName: "example.test", + Recipients: []string{"you@remote.test"}, FileSentCopy: true, + Body: bytes.NewBufferString("From: me@example.test\r\nTo: you@remote.test\r\nSubject: Aviso\r\n\r\nhi"), + } + if err := submission.Submit(context.Background(), smtp); err != nil { + t.Fatalf("smtp submit: %v", err) + } + + if len(messages.persisted) != 2 { + t.Fatalf("want one copy per path (2), got %d: the webmail must not file its own copy as well", len(messages.persisted)) + } + fromWeb, fromSMTP := messages.persisted[0], messages.persisted[1] + for _, got := range []port.PersistInboundMessageInput{fromWeb, fromSMTP} { + if got.TargetFolderName != "Sent" || !got.AlreadySeen || got.MailboxID != mailboxID { + t.Errorf("a Sent copy is not what the rule says: %+v", got) + } + if got.SenderAddress != "me@example.test" || got.RecipientAddress != "me@example.test" { + t.Errorf("a Sent copy belongs to its sender: %+v", got) + } + if got.Subject != "Aviso" { + t.Errorf("subject = %q, want Aviso", got.Subject) + } + } +} diff --git a/services/protocols/internal/application/usecase/webmail_compose_test.go b/services/protocols/internal/application/usecase/webmail_compose_test.go index 161abf0..0b7dffd 100644 --- a/services/protocols/internal/application/usecase/webmail_compose_test.go +++ b/services/protocols/internal/application/usecase/webmail_compose_test.go @@ -197,6 +197,8 @@ func newSendableUseCaseWith( } submission := NewProcessOutboundEmailUseCase( nil, &countingOutboundRepo{}, &capturingBlobStorage{}, &recordingSigner{}) + // The Sent copy is filed by the submission, as it is for SMTP clients + submission.SetSentCopies(messages) return (&WebmailUseCase{ repo: repo, auth: &stubAuthRepo{account: account}, diff --git a/services/protocols/internal/application/usecase/webmail_mail.go b/services/protocols/internal/application/usecase/webmail_mail.go index 361c9f2..f67c0af 100644 --- a/services/protocols/internal/application/usecase/webmail_mail.go +++ b/services/protocols/internal/application/usecase/webmail_mail.go @@ -32,10 +32,9 @@ type WebmailUseCase struct { submission *ProcessOutboundEmailUseCase auth port.AuthRepository localHost string - // blobStore and messages file the sender's own copy into Sent, and hold - // drafts. Both are optional: when either is nil the copy is skipped rather - // than the send failing, because a message that has already been queued - // for delivery must not be reported as an error. + // blobStore and messages hold drafts. Both are optional: when either is nil + // saving a draft reports ErrDraftsUnavailable. (The copy in Sent is filed by + // the submission use case, not here.) blobStore port.BlobStorage messages port.InboundMessageRepository } @@ -53,7 +52,7 @@ func NewWebmailUseCase( return &WebmailUseCase{repo: repo, blobs: blobs, submission: submission, auth: auth, localHost: localHost} } -// WithLocalFiling enables the Sent copy and draft storage. +// WithLocalFiling enables draft storage. func (uc *WebmailUseCase) WithLocalFiling(store port.BlobStorage, messages port.InboundMessageRepository) *WebmailUseCase { uc.blobStore = store uc.messages = messages @@ -188,6 +187,10 @@ func (uc *WebmailUseCase) Send(ctx context.Context, input ComposeInput) error { payload := buildMimeMessage(input, recipients, uc.localHost) + // The copy in Sent is filed by Submit, the same way it is for a client that + // sends over SMTP: one rule, one place. It used to be filed here, after the + // submission, and not at all for SMTP, so an account that sent by SMTP had an + // empty Sent folder. if err := uc.submission.Submit(ctx, ProcessOutboundEmailInput{ MailboxID: account.ID, SenderAddr: account.EmailAddress, @@ -195,19 +198,11 @@ func (uc *WebmailUseCase) Send(ctx context.Context, input ComposeInput) error { Recipients: recipients, Body: bytes.NewReader(payload), MaxRecipientsPerHour: account.MaxRecipientsPerHour, + FileSentCopy: true, }); err != nil { return err } - // Only after the message is durably queued. Filing it first would leave a - // copy in Sent for a message that was then rejected by the send limit. - // - // A failure here is logged and swallowed: the mail is already on its way, - // and telling the composer the send failed would invite a duplicate. - if err := uc.fileLocalCopy(ctx, account.ID, account.EmailAddress, "Sent", payload); err != nil { - log.Printf("webmail: could not file the Sent copy for %s: %v", account.EmailAddress, err) - } - // The draft this message grew out of has been superseded by the real // thing. Swallowed like the Sent copy above and for the same reason: the // mail is already queued, and failing the request now would only invite a @@ -288,36 +283,6 @@ func (uc *WebmailUseCase) Delete(ctx context.Context, address, folder string, ui } } -// fileLocalCopy stores a message the user themselves composed into one of -// their own folders. It reuses the inbound persistence path so the copy gets a -// real IMAP UID and shows up over IMAP as well as in the webmail. -func (uc *WebmailUseCase) fileLocalCopy( - ctx context.Context, mailboxID uuid.UUID, address, folder string, payload []byte, -) error { - if uc.blobStore == nil || uc.messages == nil { - return nil - } - blob, err := uc.blobStore.Store(ctx, bytes.NewReader(payload)) - if err != nil { - return err - } - headers := ExtractMessageHeaders(payload) - _, err = uc.messages.PersistAll(ctx, []port.PersistInboundMessageInput{{ - MailboxID: mailboxID, - Blob: blob, - SenderAddress: address, - RecipientAddress: address, - TargetFolderName: folder, - Subject: headers.Subject, - Snippet: headers.Snippet, - FromDisplayName: headers.FromDisplayName, - MessageIDHeader: headers.MessageID, - HasAttachments: headers.HasAttachments, - AlreadySeen: true, - }}) - return err -} - // SaveDraft files the message being composed into the Drafts folder. // // Nothing did this before: the composer displayed "draft saved automatically" diff --git a/services/protocols/internal/e2e/one_mailbox_everywhere_e2e_test.go b/services/protocols/internal/e2e/one_mailbox_everywhere_e2e_test.go new file mode 100644 index 0000000..3153434 --- /dev/null +++ b/services/protocols/internal/e2e/one_mailbox_everywhere_e2e_test.go @@ -0,0 +1,385 @@ +package e2e + +import ( + "bytes" + "context" + "crypto/tls" + "encoding/json" + "fmt" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/alexedwards/argon2id" + "github.com/emersion/go-imap/v2" + "github.com/emersion/go-imap/v2/imapclient" + "github.com/emersion/go-imap/v2/imapserver" + gosmtp "github.com/emersion/go-smtp" + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + "github.com/google/uuid" + + "lambdamail/protocols/internal/application/port" + appusecase "lambdamail/protocols/internal/application/usecase" + "lambdamail/protocols/internal/infrastructure/diskstorage" + "lambdamail/protocols/internal/infrastructure/dkim" + "lambdamail/protocols/internal/infrastructure/postgres" + httppresentation "lambdamail/protocols/internal/presentation/http" + imappresentation "lambdamail/protocols/internal/presentation/imap" + smtppresentation "lambdamail/protocols/internal/presentation/smtp" +) + +// TestOneMailboxEverywhere is the contract of the whole server: SMTP, IMAP and +// the webmail are three ways into the same mailbox, so whatever one does, the +// other two must see, with the same numbers. +// +// The webmail used to carry its own copy of the rules and its own queries, and +// the two stacks drifted apart: mail sent over SMTP never reached Sent, reading +// over IMAP never moved the unread badge, deleting in the webmail never gave the +// quota back. This test drives all three on one mailbox, mixing them, and checks +// after every step that each view agrees with the others and with the data. +func TestOneMailboxEverywhere(t *testing.T) { + if testing.Short() { + t.Skip("e2e test skipped in -short mode") + } + ctx := context.Background() + + pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig(). + Port(15451). + RuntimePath(t.TempDir()). + StartTimeout(120 * time.Second)) + if err := pg.Start(); err != nil { + t.Fatalf("start embedded postgres: %v", err) + } + defer pg.Stop() + + pool, err := postgres.NewPool(ctx, "postgres://postgres:postgres@localhost:15451/postgres?sslmode=disable") + if err != nil { + t.Fatalf("connect pool: %v", err) + } + defer pool.Close() + applyMigrations(t, ctx, pool) + + // ------------------------------------------------------------- fixtures + const ( + address = "plataforma@everywhere.test" + password = "a long enough password" + ) + hash, err := argon2id.CreateHash(password, argon2id.DefaultParams) + if err != nil { + t.Fatalf("hash: %v", err) + } + domainID, mailboxID := uuid.New(), uuid.New() + pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, 'everywhere.test', 'everywhere.test')`, domainID) + pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'plataforma', $3, $4)`, + mailboxID, domainID, address, hash) + // The standard folders come from the one definition in the database + if _, err := pool.Exec(ctx, `SELECT provision_default_folders($1)`, mailboxID); err != nil { + t.Fatalf("provision folders: %v", err) + } + + blobs := diskstorage.NewLocalDiskBlobStorage(pool, t.TempDir()) + blobReader := diskstorage.NewLocalDiskBlobReader(pool) + authRepo := postgres.NewAuthRepository(pool) + messages := postgres.NewInboundMessageRepository(pool) + + // Three messages arrive from outside, as the SMTP side would file them + inbound := appusecase.NewProcessInboundEmailUseCase(postgres.NewMailboxRepository(pool), blobs, messages) + target, err := postgres.NewMailboxRepository(pool).FindActiveByAddress(ctx, address) + if err != nil || target == nil { + t.Fatalf("resolve mailbox: %v", err) + } + for i := 1; i <= 3; i++ { + raw := fmt.Sprintf("From: alice@remote.test\r\nTo: %s\r\nSubject: inbound %d\r\nMessage-ID: \r\n\r\nbody %d\r\n", address, i, i, i) + if err := inbound.Handle(ctx, appusecase.ProcessInboundEmailInput{ + Sender: "alice@remote.test", Recipients: []port.MailboxRecord{*target}, RecipientAddresses: []string{address}, Body: strings.NewReader(raw), + }); err != nil { + t.Fatalf("deliver %d: %v", i, err) + } + } + + // -------------------------------------------- the three ways in, wired as main.go + submissionUC := appusecase.NewProcessOutboundEmailUseCase(authRepo, postgres.NewOutboundRepository(pool), blobs, dkim.NewSigner(nil)) + submissionUC.SetSentCopies(messages) + + smtpServer := gosmtp.NewServer(smtppresentation.NewSubmissionBackend(submissionUC)) + smtpServer.Domain = "mail.everywhere.test" + smtpServer.TLSConfig = &tls.Config{Certificates: []tls.Certificate{selfSignedTLSCert(t)}} + smtpLn, _ := net.Listen("tcp", "127.0.0.1:0") + defer smtpLn.Close() + go smtpServer.Serve(smtpLn) + + imapUC := appusecase.NewImapSessionUseCase( + authRepo, postgres.NewImapFolderRepository(pool), postgres.NewMessageQueryRepository(pool), + postgres.NewFlagRepository(pool), blobReader, postgres.NewExpungeRepository(pool), postgres.NewCopyRepository(pool)) + imapServer := imapserver.New(&imapserver.Options{ + NewSession: func(c *imapserver.Conn) (imapserver.Session, *imapserver.GreetingData, error) { + return imappresentation.NewSession(c, imapUC) + }, + Caps: imap.CapSet{imap.CapIMAP4rev1: {}, imap.CapMove: {}, imap.CapUIDPlus: {}}, + InsecureAuth: true, + }) + imapLn, _ := net.Listen("tcp", "127.0.0.1:0") + go imapServer.Serve(imapLn) + defer imapServer.Close() + + webmailUC := appusecase.NewWebmailUseCase(postgres.NewWebmailRepository(pool), blobReader, submissionUC, authRepo, "mail.everywhere.test"). + WithLocalFiling(blobs, messages) + router := httppresentation.NewRouter(nil, func() error { return nil }) + router.SetMailAPI(webmailUC, webmailSecret) + web := httptest.NewServer(router) + defer web.Close() + token := mintSession(t, address, "user", time.Hour) + + // --------------------------------------------------------- small drivers + webCall := func(method, path string, body any) []byte { + t.Helper() + var reader *bytes.Reader + if body != nil { + encoded, _ := json.Marshal(body) + reader = bytes.NewReader(encoded) + } else { + reader = bytes.NewReader(nil) + } + req, _ := http.NewRequest(method, web.URL+path, reader) + req.Header.Set("Cookie", "lm_user_session="+token) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("webmail %s %s: %v", method, path, err) + } + defer resp.Body.Close() + buf := new(bytes.Buffer) + buf.ReadFrom(resp.Body) + if resp.StatusCode >= 300 { + t.Fatalf("webmail %s %s = %d: %s", method, path, resp.StatusCode, buf.String()) + } + return buf.Bytes() + } + + type webFolder struct { + Name string `json:"name"` + UnreadCount int `json:"unread_count"` + TotalCount int `json:"total_count"` + } + webFolders := func() map[string]webFolder { + t.Helper() + var payload struct { + Folders []webFolder `json:"folders"` + } + raw := webCall(http.MethodGet, "/api/v1/mail/folders", nil) + if err := json.Unmarshal(raw, &payload); err != nil || payload.Folders == nil { + var plain []webFolder + if err2 := json.Unmarshal(raw, &plain); err2 != nil { + t.Fatalf("decode folders: %v / %v: %s", err, err2, raw) + } + payload.Folders = plain + } + out := map[string]webFolder{} + for _, f := range payload.Folders { + out[f.Name] = f + } + return out + } + type webMessage struct { + UID uint32 `json:"uid"` + Subject string `json:"subject"` + Seen bool `json:"seen"` + } + webMessages := func(folder string) []webMessage { + t.Helper() + var payload struct { + Messages []webMessage `json:"messages"` + } + raw := webCall(http.MethodGet, "/api/v1/mail/messages?folder="+folder, nil) + if err := json.Unmarshal(raw, &payload); err != nil || payload.Messages == nil { + var plain []webMessage + if err2 := json.Unmarshal(raw, &plain); err2 != nil { + t.Fatalf("decode messages: %v / %v: %s", err, err2, raw) + } + payload.Messages = plain + } + return payload.Messages + } + + imapClient, err := imapclient.DialInsecure(imapLn.Addr().String(), nil) + if err != nil { + t.Fatalf("dial imap: %v", err) + } + defer imapClient.Close() + if err := imapClient.Login(address, password).Wait(); err != nil { + t.Fatalf("imap login: %v", err) + } + imapExists := func(folder string) int { + t.Helper() + data, err := imapClient.Select(folder, nil).Wait() + if err != nil { + t.Fatalf("imap select %s: %v", folder, err) + } + return int(data.NumMessages) + } + imapFlags := func(folder string) map[uint32][]imap.Flag { + t.Helper() + imapExists(folder) + all := imap.UIDSet{} + all.AddRange(1, 0) // 1:* + cmd := imapClient.Fetch(all, &imap.FetchOptions{UID: true, Flags: true}) + out := map[uint32][]imap.Flag{} + msgs, err := cmd.Collect() + if err != nil { + t.Fatalf("imap fetch flags in %s: %v", folder, err) + } + for _, m := range msgs { + out[uint32(m.UID)] = m.Flags + } + return out + } + + // agree compares what the webmail and IMAP each say about every folder, and + // both with the rows: the one thing that must never differ + agree := func(after string) { + t.Helper() + assertEverywhere := webFolders() + for _, name := range []string{"INBOX", "Sent", "Trash", "Drafts"} { + if got, want := imapExists(name), assertEverywhere[name].TotalCount; got != want { + t.Errorf("after %s: %s holds %d messages over IMAP and %d in the webmail", after, name, got, want) + } + } + var badFolders, badBytes int + pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM ( + SELECT f.id FROM folders f + LEFT JOIN email_messages e ON e.folder_id = f.id AND e.expunged_at IS NULL + WHERE f.mailbox_id = $1 + GROUP BY f.id, f.total_count, f.unread_count + HAVING f.total_count <> COUNT(e.id) + OR f.unread_count <> COUNT(e.id) FILTER (WHERE NOT EXISTS ( + SELECT 1 FROM message_flags mf WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Seen')) + ) x`, mailboxID).Scan(&badFolders) + pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM mailboxes m + WHERE m.id = $1 AND m.used_bytes <> COALESCE((SELECT SUM(size_bytes) FROM email_messages WHERE mailbox_id = m.id AND expunged_at IS NULL), 0)`, + mailboxID).Scan(&badBytes) + if badFolders != 0 { + t.Errorf("after %s: %d folder(s) have cached counters that differ from their messages", after, badFolders) + } + if badBytes != 0 { + t.Errorf("after %s: the quota differs from the bytes actually stored", after) + } + } + + agree("delivery") + if f := webFolders(); f["INBOX"].UnreadCount != 3 || f["INBOX"].TotalCount != 3 { + t.Fatalf("INBOX should start with 3 unread of 3, the webmail says %+v", f["INBOX"]) + } + + // ----- 1. SMTP sends a message; Sent must show it in the webmail and over IMAP + smtpClient, err := gosmtp.DialStartTLS(smtpLn.Addr().String(), &tls.Config{InsecureSkipVerify: true}) + if err != nil { + t.Fatalf("dial smtp: %v", err) + } + if err := smtpClient.Auth(plainAuth(address, password)); err != nil { + t.Fatalf("smtp auth: %v", err) + } + smtpClient.Mail(address, nil) + smtpClient.Rcpt("bob@remote.test", nil) + w, _ := smtpClient.Data() + w.Write([]byte("From: " + address + "\r\nTo: bob@remote.test\r\nSubject: sent over smtp\r\nMessage-ID: \r\n\r\nhello\r\n")) + if err := w.Close(); err != nil { + t.Fatalf("smtp data: %v", err) + } + smtpClient.Quit() + + agree("sending over SMTP") + if sent := webMessages("sent"); len(sent) != 1 || sent[0].Subject != "sent over smtp" { + t.Fatalf("the webmail's Sent should hold the SMTP message, got %+v", sent) + } + if n := imapExists("Sent"); n != 1 { + t.Fatalf("IMAP's Sent holds %d messages after an SMTP send, want 1", n) + } + if f := webFolders(); f["Sent"].UnreadCount != 0 { + t.Errorf("the sender's own copy shows as unread: %+v", f["Sent"]) + } + + // ----- 2. the webmail sends one too: it must land in the same place, once + webCall(http.MethodPost, "/api/v1/mail/send", map[string]any{"to": []string{"carol@remote.test"}, "subject": "sent from webmail", "text": "hi"}) + agree("sending from the webmail") + if n := imapExists("Sent"); n != 2 { + t.Fatalf("Sent holds %d messages after one SMTP send and one webmail send, want exactly 2", n) + } + + // ----- 3. read in the webmail; IMAP must see \Seen and the badge must fall + inbox := webMessages("inbox") + if len(inbox) != 3 { + t.Fatalf("webmail lists %d INBOX messages, want 3", len(inbox)) + } + uidA, uidB, uidC := inbox[2].UID, inbox[1].UID, inbox[0].UID // oldest first by UID + webCall(http.MethodPost, "/api/v1/mail/seen", map[string]any{"folder": "inbox", "uid": uidA, "seen": true}) + agree("reading a message in the webmail") + if flags := imapFlags("INBOX")[uidA]; !hasFlag(flags, imap.FlagSeen) { + t.Errorf("IMAP does not see the message the webmail marked as read: %v", flags) + } + if f := webFolders(); f["INBOX"].UnreadCount != 2 { + t.Errorf("INBOX unread = %d after one read, want 2", f["INBOX"].UnreadCount) + } + + // ----- 4. read over IMAP; the webmail's badge must follow (it never did) + imapExists("INBOX") + if err := imapClient.Store(imap.UIDSetNum(imap.UID(uidB)), &imap.StoreFlags{Op: imap.StoreFlagsAdd, Silent: true, Flags: []imap.Flag{imap.FlagSeen}}, nil).Close(); err != nil { + t.Fatalf("imap store: %v", err) + } + agree("reading a message over IMAP") + if f := webFolders(); f["INBOX"].UnreadCount != 1 { + t.Errorf("INBOX unread = %d after a read over IMAP, want 1: the badge ignored it", f["INBOX"].UnreadCount) + } + for _, m := range webMessages("inbox") { + if m.UID == uidB && !m.Seen { + t.Error("the webmail lists the message IMAP marked as read as unread") + } + } + + // ----- 5. delete over IMAP (\Deleted + EXPUNGE); the webmail list shrinks, the quota is returned + imapExists("INBOX") + if err := imapClient.Store(imap.UIDSetNum(imap.UID(uidC)), &imap.StoreFlags{Op: imap.StoreFlagsAdd, Silent: true, Flags: []imap.Flag{imap.FlagDeleted}}, nil).Close(); err != nil { + t.Fatalf("imap store deleted: %v", err) + } + if err := imapClient.Expunge().Close(); err != nil { + t.Fatalf("imap expunge: %v", err) + } + agree("deleting over IMAP") + if n := len(webMessages("inbox")); n != 2 { + t.Errorf("the webmail lists %d INBOX messages after an IMAP delete, want 2", n) + } + + // ----- 6. delete in the webmail (to Trash); IMAP sees it in Trash and not in INBOX + webCall(http.MethodPost, "/api/v1/mail/delete", map[string]any{"folder": "inbox", "uid": uidA}) + agree("deleting in the webmail") + if n := imapExists("Trash"); n != 1 { + t.Errorf("IMAP's Trash holds %d messages, want 1", n) + } + if n := imapExists("INBOX"); n != 1 { + t.Errorf("IMAP's INBOX holds %d messages, want 1", n) + } + + // ----- 7. empty the Trash in the webmail: the bytes go back to the quota + trash := webMessages("trash") + if len(trash) != 1 { + t.Fatalf("the webmail lists %d Trash messages, want 1", len(trash)) + } + webCall(http.MethodPost, "/api/v1/mail/delete", map[string]any{"folder": "trash", "uid": trash[0].UID}) + agree("emptying the Trash") + if n := imapExists("Trash"); n != 0 { + t.Errorf("IMAP's Trash holds %d messages after emptying it, want 0", n) + } +} + +func hasFlag(flags []imap.Flag, want imap.Flag) bool { + for _, f := range flags { + if f == want { + return true + } + } + return false +} diff --git a/services/protocols/internal/infrastructure/postgres/copy_repository.go b/services/protocols/internal/infrastructure/postgres/copy_repository.go index 7c6482d..b98dfb3 100644 --- a/services/protocols/internal/infrastructure/postgres/copy_repository.go +++ b/services/protocols/internal/infrastructure/postgres/copy_repository.go @@ -2,133 +2,21 @@ package postgres import ( "context" - "fmt" - "github.com/google/uuid" "github.com/jackc/pgx/v5/pgxpool" "lambdamail/protocols/internal/application/port" ) -// CopyRepository implements port.CopyRepository against Postgres. One -// transaction per copied message, mirroring InboundMessageRepository's -// established per-message transactional pattern (PLAN.md section 6.1). +// CopyRepository implements port.CopyRepository (IMAP COPY) over MailStore. type CopyRepository struct { - pool *pgxpool.Pool + store *MailStore } func NewCopyRepository(pool *pgxpool.Pool) *CopyRepository { - return &CopyRepository{pool: pool} + return &CopyRepository{store: NewMailStore(pool)} } func (r *CopyRepository) CopyMessages(ctx context.Context, sourceFolderID string, uids []uint32, destFolderID string) ([]port.CopiedMessage, error) { - results := make([]port.CopiedMessage, 0, len(uids)) - for _, uid := range uids { - destUID, err := r.copyOne(ctx, sourceFolderID, uid, destFolderID) - if err != nil { - return nil, fmt.Errorf("copy uid %d: %w", uid, err) - } - results = append(results, port.CopiedMessage{SourceUID: uid, DestUID: destUID}) - } - return results, nil -} - -func (r *CopyRepository) copyOne(ctx context.Context, sourceFolderID string, sourceUID uint32, destFolderID string) (uint32, error) { - tx, err := r.pool.Begin(ctx) - if err != nil { - return 0, fmt.Errorf("begin transaction: %w", err) - } - defer tx.Rollback(ctx) - - var sourceMessageID string - var blobID, senderAddress string - var recipientAddresses []string - var sizeBytes int64 - var spfResult, dkimResult, dmarcResult *string - var fromDisplayName, subject, snippet, messageIDHeader, inReplyTo *string - var threadID *uuid.UUID - var hasAttachments bool - var arcResult, daneStatus, tlsVersion, spamVerdict, virusName *string - var spamScore float64 - err = tx.QueryRow(ctx, ` - SELECT id, blob_id, sender_address, recipient_addresses, size_bytes, spf_result, dkim_result, dmarc_result, - from_display_name, subject, snippet, message_id_header, in_reply_to, thread_id, has_attachments, - arc_result, dane_status, tls_version, spam_score, spam_verdict, virus_name - FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - `, sourceFolderID, sourceUID).Scan(&sourceMessageID, &blobID, &senderAddress, &recipientAddresses, &sizeBytes, &spfResult, &dkimResult, &dmarcResult, - &fromDisplayName, &subject, &snippet, &messageIDHeader, &inReplyTo, &threadID, &hasAttachments, - &arcResult, &daneStatus, &tlsVersion, &spamScore, &spamVerdict, &virusName) - if err != nil { - return 0, fmt.Errorf("find source message: %w", err) - } - - // uid_next is BIGINT; the destination UID we allocate for this message - // is the folder's current uid_next value before advancing it. - var destUIDNext int64 - if err := tx.QueryRow(ctx, ` - SELECT uid_next FROM folders WHERE id = $1 FOR UPDATE - `, destFolderID).Scan(&destUIDNext); err != nil { - return 0, fmt.Errorf("lock destination folder: %w", err) - } - destUID := uint32(destUIDNext) - - var destModSeq int64 - if err := tx.QueryRow(ctx, `UPDATE folders SET uid_next = uid_next + 1, highest_modseq = highest_modseq + 1, total_count = total_count + 1 WHERE id = $1 RETURNING highest_modseq`, destFolderID).Scan(&destModSeq); err != nil { - return 0, fmt.Errorf("advance destination uid_next and highest_modseq: %w", err) - } - - var wasSeen bool - if err := tx.QueryRow(ctx, ` - SELECT EXISTS (SELECT 1 FROM message_flags WHERE message_id = $1 AND flag = '\Seen') - `, sourceMessageID).Scan(&wasSeen); err != nil { - return 0, fmt.Errorf("check source seen flag: %w", err) - } - if !wasSeen { - if _, err := tx.Exec(ctx, `UPDATE folders SET unread_count = unread_count + 1 WHERE id = $1`, destFolderID); err != nil { - return 0, fmt.Errorf("increment destination unread_count: %w", err) - } - } - - var destMailboxID string - if err := tx.QueryRow(ctx, `SELECT mailbox_id FROM folders WHERE id = $1`, destFolderID).Scan(&destMailboxID); err != nil { - return 0, fmt.Errorf("find destination mailbox: %w", err) - } - - var destMessageID string - var destReceivedAt any - err = tx.QueryRow(ctx, ` - INSERT INTO email_messages ( - mailbox_id, folder_id, uid, modseq, blob_id, - sender_address, recipient_addresses, size_bytes, - spf_result, dkim_result, dmarc_result, - from_display_name, subject, snippet, message_id_header, in_reply_to, thread_id, has_attachments, - arc_result, dane_status, tls_version, spam_score, spam_verdict, virus_name - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24) - RETURNING id, received_at - `, destMailboxID, destFolderID, destUID, destModSeq, blobID, senderAddress, recipientAddresses, sizeBytes, spfResult, dkimResult, dmarcResult, - fromDisplayName, subject, snippet, messageIDHeader, inReplyTo, threadID, hasAttachments, - arcResult, daneStatus, tlsVersion, spamScore, spamVerdict, virusName).Scan(&destMessageID, &destReceivedAt) - if err != nil { - return 0, fmt.Errorf("insert copied message: %w", err) - } - - if _, err := tx.Exec(ctx, `UPDATE message_blobs SET ref_count = ref_count + 1 WHERE id = $1`, blobID); err != nil { - return 0, fmt.Errorf("increment blob ref_count: %w", err) - } - - if _, err := tx.Exec(ctx, `UPDATE mailboxes SET used_bytes = used_bytes + $1 WHERE id = $2`, sizeBytes, destMailboxID); err != nil { - return 0, fmt.Errorf("increment destination mailbox used_bytes: %w", err) - } - - if _, err := tx.Exec(ctx, ` - INSERT INTO message_flags (message_id, received_at, flag) - SELECT $1, $2, flag FROM message_flags WHERE message_id = $3 - `, destMessageID, destReceivedAt, sourceMessageID); err != nil { - return 0, fmt.Errorf("copy flags: %w", err) - } - - if err := tx.Commit(ctx); err != nil { - return 0, fmt.Errorf("commit: %w", err) - } - return destUID, nil + return r.store.Copy(ctx, sourceFolderID, uids, destFolderID) } diff --git a/services/protocols/internal/infrastructure/postgres/expunge_repository.go b/services/protocols/internal/infrastructure/postgres/expunge_repository.go index 865c207..45a56f4 100644 --- a/services/protocols/internal/infrastructure/postgres/expunge_repository.go +++ b/services/protocols/internal/infrastructure/postgres/expunge_repository.go @@ -2,91 +2,21 @@ package postgres import ( "context" - "fmt" "github.com/jackc/pgx/v5/pgxpool" ) -// ExpungeRepository implements port.ExpungeRepository against Postgres. +// ExpungeRepository implements port.ExpungeRepository (IMAP EXPUNGE). MailStore +// does the work - counters, modseq and the quota all move there, for every path. type ExpungeRepository struct { - pool *pgxpool.Pool + store *MailStore } func NewExpungeRepository(pool *pgxpool.Pool) *ExpungeRepository { - return &ExpungeRepository{pool: pool} + return &ExpungeRepository{store: NewMailStore(pool)} } -// Expunge marks each message identified by (folderID, uid) in uids as -// expunged, decrements the folder's total_count (and unread_count for any -// message that lacked \Seen), and decrements the owning mailbox's -// used_bytes by the message's size_bytes. used_bytes is logical, -// message-level accounting (mirroring how delivery and COPY add to it per -// message even when the underlying blob is deduplicated), so it is -// decremented here independently of message_blobs.ref_count, whose -// reclamation is deferred to a future GC task. One transaction per call; -// each uid is processed sequentially within it. func (r *ExpungeRepository) Expunge(ctx context.Context, folderID string, uids []uint32) error { - if len(uids) == 0 { - return nil - } - tx, err := r.pool.Begin(ctx) - if err != nil { - return fmt.Errorf("begin transaction: %w", err) - } - defer tx.Rollback(ctx) - - for _, uid := range uids { - var wasSeen bool - var mailboxID string - var sizeBytes int64 - err := tx.QueryRow(ctx, ` - SELECT EXISTS ( - SELECT 1 FROM message_flags mf - JOIN email_messages m ON m.id = mf.message_id AND m.received_at = mf.received_at - WHERE m.folder_id = $1 AND m.uid = $2 AND mf.flag = '\Seen' - ), m.mailbox_id, m.size_bytes - FROM email_messages m - WHERE m.folder_id = $1 AND m.uid = $2 - `, folderID, uid).Scan(&wasSeen, &mailboxID, &sizeBytes) - if err != nil { - return fmt.Errorf("check seen flag for uid %d: %w", uid, err) - } - - tag, err := tx.Exec(ctx, ` - UPDATE email_messages SET expunged_at = NOW() - WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - `, folderID, uid) - if err != nil { - return fmt.Errorf("expunge uid %d: %w", uid, err) - } - if tag.RowsAffected() == 0 { - // Nothing to expunge (already expunged or never existed) - the - // caller already verified \Deleted, so don't touch counters for - // a message that wasn't actually there. - continue - } - - unreadDelta := 0 - if !wasSeen { - unreadDelta = 1 - } - if _, err := tx.Exec(ctx, ` - UPDATE folders SET total_count = total_count - 1, unread_count = unread_count - $1, highest_modseq = highest_modseq + 1 - WHERE id = $2 - `, unreadDelta, folderID); err != nil { - return fmt.Errorf("decrement folder counters for uid %d: %w", uid, err) - } - - if _, err := tx.Exec(ctx, ` - UPDATE mailboxes SET used_bytes = used_bytes - $1 - WHERE id = $2 - `, sizeBytes, mailboxID); err != nil { - return fmt.Errorf("decrement mailbox used_bytes for uid %d: %w", uid, err) - } - } - - if err := tx.Commit(ctx); err != nil { - return fmt.Errorf("commit: %w", err) - } - return nil + _, err := r.store.Expunge(ctx, folderID, uids) + return err } diff --git a/services/protocols/internal/infrastructure/postgres/flag_repository.go b/services/protocols/internal/infrastructure/postgres/flag_repository.go index cedb2c1..eecd3ac 100644 --- a/services/protocols/internal/infrastructure/postgres/flag_repository.go +++ b/services/protocols/internal/infrastructure/postgres/flag_repository.go @@ -2,95 +2,23 @@ package postgres import ( "context" - "fmt" "github.com/jackc/pgx/v5/pgxpool" "lambdamail/protocols/internal/application/port" ) -// FlagRepository implements port.FlagRepository against Postgres. +// FlagRepository implements port.FlagRepository (IMAP STORE). The work is done +// by MailStore, which every path shares; this only names the message the way +// IMAP does, by folder and UID. type FlagRepository struct { - pool *pgxpool.Pool + store *MailStore } func NewFlagRepository(pool *pgxpool.Pool) *FlagRepository { - return &FlagRepository{pool: pool} + return &FlagRepository{store: NewMailStore(pool)} } func (r *FlagRepository) SetFlags(ctx context.Context, folderID string, uid uint32, op port.FlagOp, flags []string, unchangedSince uint64) (bool, error) { - tx, err := r.pool.Begin(ctx) - if err != nil { - return false, fmt.Errorf("begin transaction: %w", err) - } - defer tx.Rollback(ctx) - - var currentModSeq uint64 - if err := tx.QueryRow(ctx, ` - SELECT modseq FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - `, folderID, uid).Scan(¤tModSeq); err != nil { - return false, fmt.Errorf("find message folder=%s uid=%d: %w", folderID, uid, err) - } - - if unchangedSince > 0 && currentModSeq > unchangedSince { - return false, nil - } - - var newModSeq uint64 - if err := tx.QueryRow(ctx, ` - UPDATE folders SET highest_modseq = highest_modseq + 1 WHERE id = $1 RETURNING highest_modseq - `, folderID).Scan(&newModSeq); err != nil { - return false, fmt.Errorf("advance highest_modseq for folder %s: %w", folderID, err) - } - - if _, err := tx.Exec(ctx, ` - UPDATE email_messages SET modseq = $1 WHERE folder_id = $2 AND uid = $3 AND expunged_at IS NULL - `, newModSeq, folderID, uid); err != nil { - return false, fmt.Errorf("update message modseq: %w", err) - } - - switch op { - case port.FlagOpAdd: - for _, flag := range flags { - if _, err := tx.Exec(ctx, ` - INSERT INTO message_flags (message_id, received_at, flag) - SELECT id, received_at, $3 FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - ON CONFLICT DO NOTHING - `, folderID, uid, flag); err != nil { - return false, fmt.Errorf("add flag %q: %w", flag, err) - } - } - case port.FlagOpDel: - for _, flag := range flags { - if _, err := tx.Exec(ctx, ` - DELETE FROM message_flags WHERE (message_id, received_at) IN ( - SELECT id, received_at FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - ) AND flag = $3 - `, folderID, uid, flag); err != nil { - return false, fmt.Errorf("delete flag %q: %w", flag, err) - } - } - case port.FlagOpSet: - if _, err := tx.Exec(ctx, ` - DELETE FROM message_flags WHERE (message_id, received_at) IN ( - SELECT id, received_at FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - ) - `, folderID, uid); err != nil { - return false, fmt.Errorf("clear flags before set: %w", err) - } - for _, flag := range flags { - if _, err := tx.Exec(ctx, ` - INSERT INTO message_flags (message_id, received_at, flag) - SELECT id, received_at, $3 FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - ON CONFLICT DO NOTHING - `, folderID, uid, flag); err != nil { - return false, fmt.Errorf("set flag %q: %w", flag, err) - } - } - } - - if err := tx.Commit(ctx); err != nil { - return false, fmt.Errorf("commit: %w", err) - } - return true, nil + return r.store.SetFlags(ctx, folderID, uid, op, flags, unchangedSince) } diff --git a/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go b/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go index ee8a6b9..12ecd3a 100644 --- a/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go +++ b/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go @@ -3,57 +3,50 @@ package postgres import ( "context" - "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "lambdamail/protocols/internal/application/port" ) -// ImapFolderRepository implements port.ImapFolderRepository against Postgres. +// ImapFolderRepository resolves a mailbox's folders for LIST and SELECT. The +// question is answered by MailStore, the same one the webmail asks, so a folder +// reports the same number of messages on both. type ImapFolderRepository struct { - pool *pgxpool.Pool + store *MailStore } func NewImapFolderRepository(pool *pgxpool.Pool) *ImapFolderRepository { - return &ImapFolderRepository{pool: pool} + return &ImapFolderRepository{store: NewMailStore(pool)} +} + +func imapRecord(f FolderInfo) port.ImapFolderRecord { + return port.ImapFolderRecord{ + ID: f.ID, + Name: f.Name, + UIDNext: uint32(f.UIDNext), + UIDValidity: uint32(f.UIDValidity), + NumMessages: uint32(f.Total), + HighestModSeq: uint64(f.HighestModSeq), + } } func (r *ImapFolderRepository) FindByName(ctx context.Context, mailboxID, name string) (*port.ImapFolderRecord, error) { - row := r.pool.QueryRow(ctx, ` - SELECT f.id, f.name, f.uid_next, f.uid_validity, f.total_count, f.highest_modseq - FROM folders f - WHERE f.mailbox_id = $1 AND f.name = $2 - `, mailboxID, name) - - var rec port.ImapFolderRecord - if err := row.Scan(&rec.ID, &rec.Name, &rec.UIDNext, &rec.UIDValidity, &rec.NumMessages, &rec.HighestModSeq); err != nil { - if err == pgx.ErrNoRows { - return nil, nil - } + f, err := r.store.FolderByName(ctx, mailboxID, name) + if err != nil || f == nil { return nil, err } + rec := imapRecord(*f) return &rec, nil } func (r *ImapFolderRepository) ListFolders(ctx context.Context, mailboxID string) ([]port.ImapFolderRecord, error) { - rows, err := r.pool.Query(ctx, ` - SELECT f.id, f.name, f.uid_next, f.uid_validity, f.total_count, f.highest_modseq - FROM folders f - WHERE f.mailbox_id = $1 - ORDER BY f.name - `, mailboxID) + folders, err := r.store.Folders(ctx, mailboxID) if err != nil { return nil, err } - defer rows.Close() - var out []port.ImapFolderRecord - for rows.Next() { - var rec port.ImapFolderRecord - if err := rows.Scan(&rec.ID, &rec.Name, &rec.UIDNext, &rec.UIDValidity, &rec.NumMessages, &rec.HighestModSeq); err != nil { - return nil, err - } - out = append(out, rec) + for _, f := range folders { + out = append(out, imapRecord(f)) } - return out, rows.Err() + return out, nil } diff --git a/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go b/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go new file mode 100644 index 0000000..54f041c --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go @@ -0,0 +1,534 @@ +package postgres + +import ( + "context" + "fmt" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "lambdamail/protocols/internal/application/port" +) + +// The mailbox state contract. +// +// A message can be read, flagged, moved, copied and deleted from IMAP, from the +// webmail, and by the server itself. All of them change the same rows, and the +// rows carry cached numbers (folder counters, the folder's modseq, the +// mailbox's used bytes) that every path must keep true. Before there was one +// implementation, each path kept them true on its own, and they drifted: +// +// - reading a message over IMAP never moved the unread counter, so the badge +// in the webmail and IMAP STATUS disagreed with the flags; +// - deleting from the webmail never gave the quota back; +// - a second expunge of the same message could push a counter below zero. +// +// These tests state what must hold after any operation and run every operation +// through each path that can perform it. They do not care which code did the +// work, only that the result is the same. + +// folderState is everything observable about one folder. +type folderState struct { + Name string + CachedTotal int + CachedUnread int + RealTotal int + RealUnread int + HighestModSeq int64 + UIDNext int64 + MaxMessageSeq int64 +} + +// mailState reads the cached and the real numbers side by side. +func mailState(t *testing.T, pool *pgxpool.Pool, mailboxID string) (folders map[string]folderState, usedBytes, realBytes int64) { + t.Helper() + ctx := context.Background() + + rows, err := pool.Query(ctx, ` + SELECT f.name, f.total_count, f.unread_count, f.highest_modseq, f.uid_next, + COUNT(e.id)::int, + COUNT(e.id) FILTER (WHERE NOT EXISTS ( + SELECT 1 FROM message_flags mf + WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Seen'))::int, + COALESCE(MAX(e.modseq), 0) + FROM folders f + LEFT JOIN email_messages e ON e.folder_id = f.id AND e.expunged_at IS NULL + WHERE f.mailbox_id = $1 + GROUP BY f.id`, mailboxID) + if err != nil { + t.Fatalf("read folder state: %v", err) + } + defer rows.Close() + + folders = map[string]folderState{} + for rows.Next() { + var s folderState + if err := rows.Scan(&s.Name, &s.CachedTotal, &s.CachedUnread, &s.HighestModSeq, &s.UIDNext, &s.RealTotal, &s.RealUnread, &s.MaxMessageSeq); err != nil { + t.Fatalf("scan folder state: %v", err) + } + folders[s.Name] = s + } + + if err := pool.QueryRow(ctx, `SELECT used_bytes FROM mailboxes WHERE id = $1`, mailboxID).Scan(&usedBytes); err != nil { + t.Fatalf("read used_bytes: %v", err) + } + if err := pool.QueryRow(ctx, `SELECT COALESCE(SUM(size_bytes), 0) FROM email_messages WHERE mailbox_id = $1 AND expunged_at IS NULL`, mailboxID).Scan(&realBytes); err != nil { + t.Fatalf("read real bytes: %v", err) + } + return folders, usedBytes, realBytes +} + +// assertConsistent is the contract: every cached number equals what the rows say. +func assertConsistent(t *testing.T, pool *pgxpool.Pool, mailboxID, after string) { + t.Helper() + folders, used, real := mailState(t, pool, mailboxID) + + for name, f := range folders { + if f.CachedTotal != f.RealTotal { + t.Errorf("after %s: %s total_count = %d, but it holds %d messages", after, name, f.CachedTotal, f.RealTotal) + } + if f.CachedUnread != f.RealUnread { + t.Errorf("after %s: %s unread_count = %d, but %d messages are unread", after, name, f.CachedUnread, f.RealUnread) + } + if f.CachedUnread < 0 || f.CachedTotal < 0 { + t.Errorf("after %s: %s has a negative counter (unread %d, total %d)", after, name, f.CachedUnread, f.CachedTotal) + } + if f.HighestModSeq < f.MaxMessageSeq { + t.Errorf("after %s: %s highest_modseq %d is behind a message's modseq %d, so a CONDSTORE client would miss the change", after, name, f.HighestModSeq, f.MaxMessageSeq) + } + } + if used != real { + t.Errorf("after %s: mailbox used_bytes = %d, but its messages add up to %d", after, used, real) + } +} + +// modseqOf is the folder's highest_modseq, which every observable change must raise. +func modseqOf(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string) int64 { + t.Helper() + folders, _, _ := mailState(t, pool, mailboxID) + return folders[folder].HighestModSeq +} + +func folderID(t *testing.T, pool *pgxpool.Pool, mailboxID, name string) string { + t.Helper() + var id string + if err := pool.QueryRow(context.Background(), `SELECT id::text FROM folders WHERE mailbox_id = $1 AND name = $2`, mailboxID, name).Scan(&id); err != nil { + t.Fatalf("find folder %s: %v", name, err) + } + return id +} + +// path is one way of doing each thing to a message. IMAP and webmail both +// implement it; the contract runs against each. +type path struct { + name string + setSeen func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, seen bool) + setFlag func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, flagged bool) + expunge func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32) + moveTo func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, target string) + flagsKept func() bool +} + +func imapPath() path { + return path{ + name: "imap", + setSeen: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, seen bool) { + op := port.FlagOpAdd + if !seen { + op = port.FlagOpDel + } + if _, err := NewFlagRepository(pool).SetFlags(context.Background(), folderID(t, pool, mailboxID, folder), uid, op, []string{`\Seen`}, 0); err != nil { + t.Fatalf("imap SetFlags: %v", err) + } + }, + setFlag: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, flagged bool) { + op := port.FlagOpAdd + if !flagged { + op = port.FlagOpDel + } + if _, err := NewFlagRepository(pool).SetFlags(context.Background(), folderID(t, pool, mailboxID, folder), uid, op, []string{`\Flagged`}, 0); err != nil { + t.Fatalf("imap SetFlags: %v", err) + } + }, + expunge: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32) { + if err := NewExpungeRepository(pool).Expunge(context.Background(), folderID(t, pool, mailboxID, folder), []uint32{uid}); err != nil { + t.Fatalf("imap Expunge: %v", err) + } + }, + moveTo: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, target string) { + // IMAP has no MOVE here: a client copies, marks \Deleted and expunges + src, dst := folderID(t, pool, mailboxID, folder), folderID(t, pool, mailboxID, target) + if _, err := NewCopyRepository(pool).CopyMessages(context.Background(), src, []uint32{uid}, dst); err != nil { + t.Fatalf("imap Copy: %v", err) + } + if err := NewExpungeRepository(pool).Expunge(context.Background(), src, []uint32{uid}); err != nil { + t.Fatalf("imap Expunge after copy: %v", err) + } + }, + } +} + +func webmailPath() path { + return path{ + name: "webmail", + setSeen: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, seen bool) { + if err := NewWebmailRepository(pool).MarkSeen(context.Background(), mailboxID, folder, uid, seen); err != nil { + t.Fatalf("webmail MarkSeen: %v", err) + } + }, + setFlag: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, flagged bool) { + if err := NewWebmailRepository(pool).MarkFlagged(context.Background(), mailboxID, folder, uid, flagged); err != nil { + t.Fatalf("webmail MarkFlagged: %v", err) + } + }, + expunge: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32) { + if err := NewWebmailRepository(pool).Expunge(context.Background(), mailboxID, folder, uid); err != nil { + t.Fatalf("webmail Expunge: %v", err) + } + }, + moveTo: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, target string) { + if _, err := NewWebmailRepository(pool).MoveToFolder(context.Background(), mailboxID, folder, uid, target); err != nil { + t.Fatalf("webmail Move: %v", err) + } + }, + } +} + +func eachPath(t *testing.T, scenario func(t *testing.T, p path)) { + for _, p := range []path{imapPath(), webmailPath()} { + p := p + t.Run(p.name, func(t *testing.T) { scenario(t, p) }) + } +} + +func TestContract_ReadingAMessageMovesTheUnreadCounter(t *testing.T) { + pool := testPool(t) + eachPath(t, func(t *testing.T, p path) { + mailboxID := seedMailbox(t, pool) + a := deliver(t, pool, mailboxID, "INBOX", "a", false) + deliver(t, pool, mailboxID, "INBOX", "b", false) + assertConsistent(t, pool, mailboxID, "delivery") + + before := modseqOf(t, pool, mailboxID, "INBOX") + p.setSeen(t, pool, mailboxID, "INBOX", a, true) + assertConsistent(t, pool, mailboxID, "marking one message read") + if after := modseqOf(t, pool, mailboxID, "INBOX"); after <= before { + t.Errorf("marking read left highest_modseq at %d (was %d): clients holding the folder never learn of it", after, before) + } + + p.setSeen(t, pool, mailboxID, "INBOX", a, true) + assertConsistent(t, pool, mailboxID, "marking the same message read twice") + + p.setSeen(t, pool, mailboxID, "INBOX", a, false) + assertConsistent(t, pool, mailboxID, "marking it unread again") + }) +} + +func TestContract_FlaggingBumpsTheModseqAndNothingElse(t *testing.T) { + pool := testPool(t) + eachPath(t, func(t *testing.T, p path) { + mailboxID := seedMailbox(t, pool) + a := deliver(t, pool, mailboxID, "INBOX", "a", false) + + before := modseqOf(t, pool, mailboxID, "INBOX") + p.setFlag(t, pool, mailboxID, "INBOX", a, true) + assertConsistent(t, pool, mailboxID, "flagging") + if after := modseqOf(t, pool, mailboxID, "INBOX"); after <= before { + t.Errorf("flagging left highest_modseq at %d (was %d)", after, before) + } + p.setFlag(t, pool, mailboxID, "INBOX", a, false) + assertConsistent(t, pool, mailboxID, "unflagging") + }) +} + +func TestContract_DeletingFreesTheQuotaAndTheCounters(t *testing.T) { + pool := testPool(t) + eachPath(t, func(t *testing.T, p path) { + mailboxID := seedMailbox(t, pool) + unread := deliver(t, pool, mailboxID, "INBOX", "unread", false) + read := deliver(t, pool, mailboxID, "INBOX", "read", true) + assertConsistent(t, pool, mailboxID, "delivery") + + p.expunge(t, pool, mailboxID, "INBOX", unread) + assertConsistent(t, pool, mailboxID, "deleting an unread message") + + p.expunge(t, pool, mailboxID, "INBOX", read) + assertConsistent(t, pool, mailboxID, "deleting a read message") + + _, used, _ := mailState(t, pool, mailboxID) + if used != 0 { + t.Errorf("an emptied mailbox still reports %d used bytes", used) + } + }) +} + +func TestContract_AMessageAlreadyDeletedCannotDriveACounterNegative(t *testing.T) { + pool := testPool(t) + eachPath(t, func(t *testing.T, p path) { + mailboxID := seedMailbox(t, pool) + a := deliver(t, pool, mailboxID, "INBOX", "a", false) + deliver(t, pool, mailboxID, "INBOX", "b", false) + + p.expunge(t, pool, mailboxID, "INBOX", a) + // The second one is the same message again. The webmail reports it as + // not found; IMAP ignores it. Either way no number may move. + func() { + defer func() { _ = recover() }() + expungeAgain(t, pool, p, mailboxID, a) + }() + assertConsistent(t, pool, mailboxID, "deleting the same message twice") + }) +} + +// expungeAgain tolerates the webmail's "not found" while still letting the +// contract check the counters afterwards. +func expungeAgain(t *testing.T, pool *pgxpool.Pool, p path, mailboxID string, uid uint32) { + t.Helper() + switch p.name { + case "imap": + _ = NewExpungeRepository(pool).Expunge(context.Background(), folderID(t, pool, mailboxID, "INBOX"), []uint32{uid}) + default: + _ = NewWebmailRepository(pool).Expunge(context.Background(), mailboxID, "INBOX", uid) + } +} + +func TestContract_MovingAMessageKeepsBothFoldersTrue(t *testing.T) { + pool := testPool(t) + eachPath(t, func(t *testing.T, p path) { + mailboxID := seedMailbox(t, pool) + unread := deliver(t, pool, mailboxID, "INBOX", "unread", false) + read := deliver(t, pool, mailboxID, "INBOX", "read", true) + + p.moveTo(t, pool, mailboxID, "INBOX", unread, "Trash") + assertConsistent(t, pool, mailboxID, "moving an unread message") + p.moveTo(t, pool, mailboxID, "INBOX", read, "Trash") + assertConsistent(t, pool, mailboxID, "moving a read message") + + folders, _, _ := mailState(t, pool, mailboxID) + if folders["INBOX"].RealTotal != 0 || folders["Trash"].RealTotal != 2 { + t.Errorf("INBOX holds %d and Trash %d, want 0 and 2", folders["INBOX"].RealTotal, folders["Trash"].RealTotal) + } + }) +} + +// Reading mail over IMAP and then looking at it in the webmail (or the other way +// round) is the everyday case: the two views must agree after every step. +func TestContract_ImapAndWebmailAgreeOnTheSameMailbox(t *testing.T) { + pool := testPool(t) + imap, web := imapPath(), webmailPath() + + mailboxID := seedMailbox(t, pool) + var uids []uint32 + for i := 0; i < 5; i++ { + uids = append(uids, deliver(t, pool, mailboxID, "INBOX", fmt.Sprintf("m%d", i), false)) + } + + imap.setSeen(t, pool, mailboxID, "INBOX", uids[0], true) + assertConsistent(t, pool, mailboxID, "reading m0 over IMAP") + web.setSeen(t, pool, mailboxID, "INBOX", uids[1], true) + assertConsistent(t, pool, mailboxID, "reading m1 in the webmail") + imap.setFlag(t, pool, mailboxID, "INBOX", uids[2], true) + web.expunge(t, pool, mailboxID, "INBOX", uids[3]) + assertConsistent(t, pool, mailboxID, "deleting m3 in the webmail") + imap.expunge(t, pool, mailboxID, "INBOX", uids[1]) + assertConsistent(t, pool, mailboxID, "deleting the already-read m1 over IMAP") + web.moveTo(t, pool, mailboxID, "INBOX", uids[4], "Trash") + assertConsistent(t, pool, mailboxID, "moving m4 to Trash in the webmail") + imap.moveTo(t, pool, mailboxID, "INBOX", uids[2], "Trash") + assertConsistent(t, pool, mailboxID, "moving m2 to Trash over IMAP") + + // What the webmail reports for a folder is what IMAP would answer + ctx := context.Background() + wmFolders, err := NewWebmailRepository(pool).ListFolders(ctx, mailboxID) + if err != nil { + t.Fatalf("webmail ListFolders: %v", err) + } + imFolders, err := NewImapFolderRepository(pool).ListFolders(ctx, mailboxID) + if err != nil { + t.Fatalf("imap ListFolders: %v", err) + } + imapTotals := map[string]int{} + for _, f := range imFolders { + imapTotals[f.Name] = int(f.NumMessages) + } + for _, f := range wmFolders { + if got := imapTotals[f.Name]; got != f.TotalCount { + t.Errorf("%s: the webmail shows %d messages and IMAP says %d", f.Name, f.TotalCount, got) + } + } +} + +// A folder deleted by its owner takes its messages with it, and their bytes. +func TestContract_DeletingAFolderReturnsItsBytes(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + mailboxID := seedMailbox(t, pool) + repo := NewWebmailRepository(pool) + + if err := repo.CreateFolder(ctx, mailboxID, "Projetos"); err != nil { + t.Fatalf("create folder: %v", err) + } + deliver(t, pool, mailboxID, "Projetos", "one", false) + deliver(t, pool, mailboxID, "Projetos", "two", true) + deliver(t, pool, mailboxID, "INBOX", "stays", false) + assertConsistent(t, pool, mailboxID, "filing into a new folder") + + if err := repo.DeleteFolder(ctx, mailboxID, "Projetos"); err != nil { + t.Fatalf("delete folder: %v", err) + } + assertConsistent(t, pool, mailboxID, "deleting the folder") +} + +// IMAP's STORE can replace a message's whole flag set; the unread counter has to +// follow the \Seen that comes or goes with it. +func TestContract_ReplacingTheFlagSetKeepsTheUnreadCounterTrue(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + mailboxID := seedMailbox(t, pool) + a := deliver(t, pool, mailboxID, "INBOX", "a", false) + b := deliver(t, pool, mailboxID, "INBOX", "b", true) + inbox := folderID(t, pool, mailboxID, "INBOX") + repo := NewFlagRepository(pool) + + if _, err := repo.SetFlags(ctx, inbox, a, port.FlagOpSet, []string{`\Seen`, `\Flagged`}, 0); err != nil { + t.Fatalf("SetFlags: %v", err) + } + assertConsistent(t, pool, mailboxID, "replacing flags with one that includes \\Seen") + + if _, err := repo.SetFlags(ctx, inbox, b, port.FlagOpSet, []string{`\Flagged`}, 0); err != nil { + t.Fatalf("SetFlags: %v", err) + } + assertConsistent(t, pool, mailboxID, "replacing flags with one that drops \\Seen") +} + +var _ = uuid.Nil + +// IMAP clients, the webmail and the delivery path all touch a mailbox at the same +// moment. The counters are shared rows, so they have to stay true under that. +func TestContract_ConcurrentChangesFromBothPathsKeepTheCountersTrue(t *testing.T) { + pool := testPool(t) + mailboxID := seedMailbox(t, pool) + + const n = 24 + uids := make([]uint32, 0, n) + for i := 0; i < n; i++ { + uids = append(uids, deliver(t, pool, mailboxID, "INBOX", fmt.Sprintf("m%d", i), false)) + } + + imap, web := imapPath(), webmailPath() + done := make(chan struct{}, n) + for i, uid := range uids { + go func(i int, uid uint32) { + defer func() { done <- struct{}{} }() + switch i % 3 { + case 0: + imap.setSeen(t, pool, mailboxID, "INBOX", uid, true) + case 1: + web.setSeen(t, pool, mailboxID, "INBOX", uid, true) + default: + if i%2 == 0 { + imap.expunge(t, pool, mailboxID, "INBOX", uid) + } else { + web.expunge(t, pool, mailboxID, "INBOX", uid) + } + } + }(i, uid) + } + for range uids { + <-done + } + + assertConsistent(t, pool, mailboxID, "24 concurrent changes from IMAP and the webmail") +} + +// Two clients deleting the same message at once is the realistic race (the +// webmail and a phone both open on the same mail). It is removed once, and every +// number moves once. +func TestContract_DeletingTheSameMessageConcurrentlyCountsOnce(t *testing.T) { + pool := testPool(t) + mailboxID := seedMailbox(t, pool) + target := deliver(t, pool, mailboxID, "INBOX", "contested", false) + deliver(t, pool, mailboxID, "INBOX", "bystander", false) + inbox := folderID(t, pool, mailboxID, "INBOX") + + ctx := context.Background() + store := NewMailStore(pool) + removed := make(chan int, 12) + for i := 0; i < 12; i++ { + go func() { + n, err := store.Expunge(ctx, inbox, []uint32{target}) + if err != nil { + t.Errorf("expunge: %v", err) + } + removed <- n + }() + } + total := 0 + for i := 0; i < 12; i++ { + total += <-removed + } + + if total != 1 { + t.Errorf("12 concurrent expunges of one message removed it %d times, want exactly once", total) + } + assertConsistent(t, pool, mailboxID, "12 clients deleting the same message") + folders, used, _ := mailState(t, pool, mailboxID) + if folders["INBOX"].RealTotal != 1 || used != 10 { + t.Errorf("INBOX holds %d and the mailbox %d bytes, want 1 message and 10 bytes", folders["INBOX"].RealTotal, used) + } +} + +// Moves in opposite directions, flag changes and deletes, all at once on the same +// two folders. Locking the folders and the message in different orders in +// different operations deadlocks here; a fixed order does not. +func TestContract_OppositeMovesAndFlagChangesNeverDeadlock(t *testing.T) { + pool := testPool(t) + mailboxID := seedMailbox(t, pool) + + const each = 20 + var inboxUIDs, trashUIDs []uint32 + for i := 0; i < each; i++ { + inboxUIDs = append(inboxUIDs, deliver(t, pool, mailboxID, "INBOX", fmt.Sprintf("i%d", i), false)) + trashUIDs = append(trashUIDs, deliver(t, pool, mailboxID, "Trash", fmt.Sprintf("t%d", i), false)) + } + + inbox, trash := folderID(t, pool, mailboxID, "INBOX"), folderID(t, pool, mailboxID, "Trash") + store := NewMailStore(pool) + ctx := context.Background() + + errs := make(chan error, each*4) + var pending int + run := func(f func() error) { + pending++ + go func() { errs <- f() }() + } + for i := 0; i < each; i++ { + i := i + run(func() error { _, err := store.Move(ctx, inbox, inboxUIDs[i], trash); return nil2(err) }) + run(func() error { _, err := store.Move(ctx, trash, trashUIDs[i], inbox); return nil2(err) }) + // a flag change on a message that is being moved elsewhere, and a delete + run(func() error { + _, err := store.SetFlags(ctx, inbox, inboxUIDs[(i+1)%each], port.FlagOpAdd, []string{`\Flagged`}, 0) + return nil2(err) + }) + run(func() error { _, err := store.Expunge(ctx, trash, []uint32{trashUIDs[(i+2)%each]}); return nil2(err) }) + } + for ; pending > 0; pending-- { + if err := <-errs; err != nil { + t.Errorf("a concurrent operation failed (a deadlock, if it says 40P01): %v", err) + } + } + + assertConsistent(t, pool, mailboxID, "opposite moves, flag changes and deletes all at once") +} + +// nil2 lets the contract tolerate "the message was moved or deleted first", a +// legitimate outcome of a race, while still surfacing every other error (above +// all a deadlock, SQLSTATE 40P01). +func nil2(err error) error { + if err == ErrMessageNotFound { + return nil + } + return err +} diff --git a/services/protocols/internal/infrastructure/postgres/mail_store.go b/services/protocols/internal/infrastructure/postgres/mail_store.go new file mode 100644 index 0000000..1a0fb30 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/mail_store.go @@ -0,0 +1,486 @@ +package postgres + +import ( + "context" + "errors" + "fmt" + "sort" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "lambdamail/protocols/internal/application/port" +) + +// MailStore is the one place that changes the state of messages and folders. +// +// IMAP, POP3, the webmail and the server's own filing all end up changing the +// same rows, and those rows carry cached numbers that have to stay true: the +// folder's unread and total counters, its highest modseq (what a CONDSTORE +// client uses to learn that something changed), the next UID, and the +// mailbox's used bytes. When each path implemented this on its own they +// drifted apart - reading over IMAP never moved the unread counter, deleting in +// the webmail never gave the quota back, a repeated expunge could push a +// counter below zero - and every fix had to be made twice, which is how one of +// the two went missing. +// +// So each state change is written here, once, in one transaction, and every +// path calls it. The repositories IMAP and the webmail used are now thin +// adapters over this: they translate their way of naming things (a folder id +// and a UID, or a mailbox, a folder name and a UID) and nothing else. +// +// Rules held for every method: +// +// - A counter never goes below zero, so a double operation cannot corrupt it. +// - An operation that changes what a client can observe raises the folder's +// modseq; one that changes nothing does not. +// - A message is located by folder, so another mailbox's mail cannot be +// reached by guessing a UID. +type MailStore struct { + pool *pgxpool.Pool +} + +func NewMailStore(pool *pgxpool.Pool) *MailStore { + return &MailStore{pool: pool} +} + +// lockFolders takes the row lock of each folder, in a fixed order. +// +// Every change updates counters on a folder row, and a move updates two. If one +// operation locked the destination folder and then the source, and another did +// the reverse, two moves in opposite directions deadlocked, and so did a move +// racing a flag change (it held the message and wanted the folder, the move held +// the folder and wanted the message). So the order is the same everywhere: +// folders first, lowest id first, then the message. +func lockFolders(ctx context.Context, tx pgx.Tx, ids ...string) error { + sort.Strings(ids) + for i, id := range ids { + if i > 0 && id == ids[i-1] { + continue + } + if _, err := tx.Exec(ctx, `SELECT 1 FROM folders WHERE id = $1 FOR UPDATE`, id); err != nil { + return fmt.Errorf("lock folder %s: %w", id, err) + } + } + return nil +} + +// messageRow is the slice of a message every operation needs. +type messageRow struct { + ID uuid.UUID + ReceivedAt any + MailboxID string + Size int64 + ModSeq uint64 + Seen bool +} + +// loadMessage finds a live message by folder and UID and locks it, so two +// concurrent changes to the same message cannot both read the old state. +func loadMessage(ctx context.Context, tx pgx.Tx, folderID string, uid uint32) (*messageRow, error) { + var m messageRow + err := tx.QueryRow(ctx, ` + SELECT e.id, e.received_at, e.mailbox_id::text, e.size_bytes, e.modseq, + EXISTS (SELECT 1 FROM message_flags mf + WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Seen') + FROM email_messages e + WHERE e.folder_id = $1 AND e.uid = $2 AND e.expunged_at IS NULL + FOR UPDATE + `, folderID, int64(uid)).Scan(&m.ID, &m.ReceivedAt, &m.MailboxID, &m.Size, &m.ModSeq, &m.Seen) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrMessageNotFound + } + if err != nil { + return nil, fmt.Errorf("find message folder=%s uid=%d: %w", folderID, uid, err) + } + return &m, nil +} + +// ResolveFolder finds a folder of a mailbox by its special-use role or by its +// name, the way the webmail and IMAP clients both address one ("trash" and +// "Trash" reach the same place; a folder the user made has no role and is +// reached by name). port.ErrFolderMissing when there is none. +func (s *MailStore) ResolveFolder(ctx context.Context, mailboxID, ref string) (string, error) { + var id string + err := s.pool.QueryRow(ctx, ` + SELECT id::text FROM folders + WHERE mailbox_id = $1 AND (special_use = LOWER($2) OR LOWER(name) = LOWER($2)) + ORDER BY special_use IS NOT NULL DESC + LIMIT 1 + `, mailboxID, ref).Scan(&id) + if errors.Is(err, pgx.ErrNoRows) { + return "", port.ErrFolderMissing + } + return id, err +} + +// SetFlags applies op to one message's flags. Add and Del touch individual +// flags, Set replaces the whole set. If unchangedSince > 0 and the message has +// changed since, nothing is written and updated is false (RFC 7162 +// UNCHANGEDSINCE). +// +// \Seen is the flag with a side effect: the folder's unread counter follows it, +// in the same transaction, by looking at the flag before and after rather than +// at which operation was asked for. That is what keeps STORE +FLAGS, STORE +// -FLAGS, STORE FLAGS (replace) and the webmail's mark-as-read all counting the +// same way. +func (s *MailStore) SetFlags(ctx context.Context, folderID string, uid uint32, op port.FlagOp, flags []string, unchangedSince uint64) (bool, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return false, fmt.Errorf("begin transaction: %w", err) + } + defer tx.Rollback(ctx) + + if err := lockFolders(ctx, tx, folderID); err != nil { + return false, err + } + msg, err := loadMessage(ctx, tx, folderID, uid) + if err != nil { + return false, err + } + + if unchangedSince > 0 && msg.ModSeq > unchangedSince { + return false, nil + } + + before, err := flagSet(ctx, tx, msg) + if err != nil { + return false, err + } + + switch op { + case port.FlagOpAdd: + for _, flag := range flags { + if _, err := tx.Exec(ctx, ` + INSERT INTO message_flags (message_id, received_at, flag) VALUES ($1, $2, $3) + ON CONFLICT DO NOTHING`, msg.ID, msg.ReceivedAt, flag); err != nil { + return false, fmt.Errorf("add flag %q: %w", flag, err) + } + } + case port.FlagOpDel: + for _, flag := range flags { + if _, err := tx.Exec(ctx, ` + DELETE FROM message_flags WHERE message_id = $1 AND received_at = $2 AND flag = $3`, + msg.ID, msg.ReceivedAt, flag); err != nil { + return false, fmt.Errorf("delete flag %q: %w", flag, err) + } + } + case port.FlagOpSet: + if _, err := tx.Exec(ctx, ` + DELETE FROM message_flags WHERE message_id = $1 AND received_at = $2`, msg.ID, msg.ReceivedAt); err != nil { + return false, fmt.Errorf("clear flags before set: %w", err) + } + for _, flag := range flags { + if _, err := tx.Exec(ctx, ` + INSERT INTO message_flags (message_id, received_at, flag) VALUES ($1, $2, $3) + ON CONFLICT DO NOTHING`, msg.ID, msg.ReceivedAt, flag); err != nil { + return false, fmt.Errorf("set flag %q: %w", flag, err) + } + } + } + + after, err := flagSet(ctx, tx, msg) + if err != nil { + return false, err + } + + // Nothing changed: no modseq, no counter. A client that stores a flag the + // message already has must not be told the folder changed. + if sameFlags(before, after) { + return true, tx.Commit(ctx) + } + + unreadDelta := 0 + switch { + case before[`\Seen`] && !after[`\Seen`]: + unreadDelta = 1 + case !before[`\Seen`] && after[`\Seen`]: + unreadDelta = -1 + } + + var modseq int64 + if err := tx.QueryRow(ctx, ` + UPDATE folders + SET highest_modseq = highest_modseq + 1, + unread_count = GREATEST(0, unread_count + $2) + WHERE id = $1 + RETURNING highest_modseq`, folderID, unreadDelta).Scan(&modseq); err != nil { + return false, fmt.Errorf("advance folder modseq: %w", err) + } + if _, err := tx.Exec(ctx, `UPDATE email_messages SET modseq = $3 WHERE id = $1 AND received_at = $2`, + msg.ID, msg.ReceivedAt, modseq); err != nil { + return false, fmt.Errorf("update message modseq: %w", err) + } + + return true, tx.Commit(ctx) +} + +func flagSet(ctx context.Context, tx pgx.Tx, msg *messageRow) (map[string]bool, error) { + rows, err := tx.Query(ctx, `SELECT flag FROM message_flags WHERE message_id = $1 AND received_at = $2`, msg.ID, msg.ReceivedAt) + if err != nil { + return nil, fmt.Errorf("read flags: %w", err) + } + defer rows.Close() + set := map[string]bool{} + for rows.Next() { + var f string + if err := rows.Scan(&f); err != nil { + return nil, err + } + set[f] = true + } + return set, rows.Err() +} + +func sameFlags(a, b map[string]bool) bool { + if len(a) != len(b) { + return false + } + for k := range a { + if !b[k] { + return false + } + } + return true +} + +// Expunge removes messages from a folder (soft delete: the row stays, marked, +// and the blob stays referenced until the retention sweep, so a mistake is +// recoverable and a client holding the UID still resolves it). +// +// For each message that was really there it lowers the folder's total, lowers +// unread if the message was unread, and gives its bytes back to the mailbox's +// quota. A message that is already gone is skipped, touching nothing, which is +// what makes a repeated expunge safe. It returns how many messages were +// removed. +func (s *MailStore) Expunge(ctx context.Context, folderID string, uids []uint32) (int, error) { + if len(uids) == 0 { + return 0, nil + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return 0, fmt.Errorf("begin transaction: %w", err) + } + defer tx.Rollback(ctx) + + if err := lockFolders(ctx, tx, folderID); err != nil { + return 0, err + } + + removed := 0 + for _, uid := range uids { + msg, err := loadMessage(ctx, tx, folderID, uid) + if errors.Is(err, ErrMessageNotFound) { + continue + } + if err != nil { + return 0, err + } + + if _, err := tx.Exec(ctx, `UPDATE email_messages SET expunged_at = NOW() WHERE id = $1 AND received_at = $2`, + msg.ID, msg.ReceivedAt); err != nil { + return 0, fmt.Errorf("expunge uid %d: %w", uid, err) + } + + unreadDelta := 0 + if !msg.Seen { + unreadDelta = -1 + } + if _, err := tx.Exec(ctx, ` + UPDATE folders + SET total_count = GREATEST(0, total_count - 1), + unread_count = GREATEST(0, unread_count + $2), + highest_modseq = highest_modseq + 1 + WHERE id = $1`, folderID, unreadDelta); err != nil { + return 0, fmt.Errorf("lower folder counters for uid %d: %w", uid, err) + } + + if _, err := tx.Exec(ctx, `UPDATE mailboxes SET used_bytes = GREATEST(0, used_bytes - $1) WHERE id = $2`, + msg.Size, msg.MailboxID); err != nil { + return 0, fmt.Errorf("give quota back for uid %d: %w", uid, err) + } + removed++ + } + + return removed, tx.Commit(ctx) +} + +// Move relocates one message to another folder of the same mailbox and returns +// the UID it has there. The message keeps its identity, its flags and its blob; +// only the folder and the UID change, because a UID means something only inside +// one folder. The quota does not move: the bytes stay in the same mailbox. +func (s *MailStore) Move(ctx context.Context, sourceFolderID string, uid uint32, destFolderID string) (uint32, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return 0, err + } + defer tx.Rollback(ctx) + + if err := lockFolders(ctx, tx, sourceFolderID, destFolderID); err != nil { + return 0, err + } + msg, err := loadMessage(ctx, tx, sourceFolderID, uid) + if err != nil { + return 0, err + } + + var destUID int64 + if err := tx.QueryRow(ctx, `SELECT uid_next FROM folders WHERE id = $1`, destFolderID).Scan(&destUID); err != nil { + return 0, fmt.Errorf("read destination uid: %w", err) + } + + unread := 0 + if !msg.Seen { + unread = 1 + } + + var modseq int64 + if err := tx.QueryRow(ctx, ` + UPDATE folders + SET uid_next = uid_next + 1, highest_modseq = highest_modseq + 1, + total_count = total_count + 1, unread_count = unread_count + $2 + WHERE id = $1 + RETURNING highest_modseq`, destFolderID, unread).Scan(&modseq); err != nil { + return 0, fmt.Errorf("advance destination folder: %w", err) + } + + if _, err := tx.Exec(ctx, `UPDATE email_messages SET folder_id = $3, uid = $4, modseq = $5 WHERE id = $1 AND received_at = $2`, + msg.ID, msg.ReceivedAt, destFolderID, destUID, modseq); err != nil { + return 0, fmt.Errorf("move message: %w", err) + } + + if _, err := tx.Exec(ctx, ` + UPDATE folders + SET total_count = GREATEST(0, total_count - 1), + unread_count = GREATEST(0, unread_count - $2), + highest_modseq = highest_modseq + 1 + WHERE id = $1`, sourceFolderID, unread); err != nil { + return 0, fmt.Errorf("lower source folder counters: %w", err) + } + + return uint32(destUID), tx.Commit(ctx) +} + +// Copy duplicates messages into another folder. Each copy is its own message +// (own UID, own flags copied from the original) that shares the original's +// blob, and it counts against the quota like any other message. One transaction +// per message, so a failure part-way leaves the earlier copies whole. +func (s *MailStore) Copy(ctx context.Context, sourceFolderID string, uids []uint32, destFolderID string) ([]port.CopiedMessage, error) { + results := make([]port.CopiedMessage, 0, len(uids)) + for _, uid := range uids { + destUID, err := s.copyOne(ctx, sourceFolderID, uid, destFolderID) + if err != nil { + return nil, fmt.Errorf("copy uid %d: %w", uid, err) + } + results = append(results, port.CopiedMessage{SourceUID: uid, DestUID: destUID}) + } + return results, nil +} + +func (s *MailStore) copyOne(ctx context.Context, sourceFolderID string, sourceUID uint32, destFolderID string) (uint32, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return 0, fmt.Errorf("begin transaction: %w", err) + } + defer tx.Rollback(ctx) + + if err := lockFolders(ctx, tx, destFolderID); err != nil { + return 0, err + } + src, err := loadMessage(ctx, tx, sourceFolderID, sourceUID) + if err != nil { + return 0, err + } + + var destUID int64 + var destMailboxID string + if err := tx.QueryRow(ctx, `SELECT uid_next, mailbox_id::text FROM folders WHERE id = $1`, destFolderID).Scan(&destUID, &destMailboxID); err != nil { + return 0, fmt.Errorf("read destination folder: %w", err) + } + + unread := 0 + if !src.Seen { + unread = 1 + } + var modseq int64 + if err := tx.QueryRow(ctx, ` + UPDATE folders + SET uid_next = uid_next + 1, highest_modseq = highest_modseq + 1, + total_count = total_count + 1, unread_count = unread_count + $2 + WHERE id = $1 + RETURNING highest_modseq`, destFolderID, unread).Scan(&modseq); err != nil { + return 0, fmt.Errorf("advance destination folder: %w", err) + } + + var newID uuid.UUID + var newReceivedAt any + if err := tx.QueryRow(ctx, ` + INSERT INTO email_messages ( + mailbox_id, folder_id, uid, modseq, blob_id, + sender_address, recipient_addresses, size_bytes, + spf_result, dkim_result, dmarc_result, + from_display_name, subject, snippet, message_id_header, in_reply_to, thread_id, has_attachments, + arc_result, dane_status, tls_version, spam_score, spam_verdict, virus_name + ) + SELECT $2::uuid, $3::uuid, $4::bigint, $5::bigint, blob_id, + sender_address, recipient_addresses, size_bytes, + spf_result, dkim_result, dmarc_result, + from_display_name, subject, snippet, message_id_header, in_reply_to, thread_id, has_attachments, + arc_result, dane_status, tls_version, spam_score, spam_verdict, virus_name + FROM email_messages WHERE id = $1 AND received_at = $6 + RETURNING id, received_at`, + src.ID, destMailboxID, destFolderID, destUID, modseq, src.ReceivedAt).Scan(&newID, &newReceivedAt); err != nil { + return 0, fmt.Errorf("insert copied message: %w", err) + } + + if _, err := tx.Exec(ctx, `UPDATE message_blobs SET ref_count = ref_count + 1 WHERE id = (SELECT blob_id FROM email_messages WHERE id = $1 AND received_at = $2)`, + newID, newReceivedAt); err != nil { + return 0, fmt.Errorf("share the blob: %w", err) + } + if _, err := tx.Exec(ctx, `UPDATE mailboxes SET used_bytes = used_bytes + $1 WHERE id = $2`, src.Size, destMailboxID); err != nil { + return 0, fmt.Errorf("count the copy against the quota: %w", err) + } + if _, err := tx.Exec(ctx, ` + INSERT INTO message_flags (message_id, received_at, flag) + SELECT $1, $2, flag FROM message_flags WHERE message_id = $3 AND received_at = $4`, + newID, newReceivedAt, src.ID, src.ReceivedAt); err != nil { + return 0, fmt.Errorf("copy flags: %w", err) + } + + return uint32(destUID), tx.Commit(ctx) +} + +// DeleteFolder removes a folder the user made, with the mail in it. The +// messages are expunged rather than deleted (recoverable, like any expunge) and +// their bytes go back to the quota. +func (s *MailStore) DeleteFolder(ctx context.Context, folderID string) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(ctx) + + var mailboxID string + if err := tx.QueryRow(ctx, `SELECT mailbox_id::text FROM folders WHERE id = $1 FOR UPDATE`, folderID).Scan(&mailboxID); err != nil { + return fmt.Errorf("lock folder: %w", err) + } + + var freed int64 + if err := tx.QueryRow(ctx, ` + WITH gone AS ( + UPDATE email_messages SET expunged_at = NOW() + WHERE folder_id = $1 AND expunged_at IS NULL + RETURNING size_bytes) + SELECT COALESCE(SUM(size_bytes), 0) FROM gone`, folderID).Scan(&freed); err != nil { + return fmt.Errorf("expunge the folder's messages: %w", err) + } + + if _, err := tx.Exec(ctx, `UPDATE mailboxes SET used_bytes = GREATEST(0, used_bytes - $1) WHERE id = $2`, freed, mailboxID); err != nil { + return fmt.Errorf("give quota back: %w", err) + } + if _, err := tx.Exec(ctx, `DELETE FROM folders WHERE id = $1`, folderID); err != nil { + return fmt.Errorf("delete folder: %w", err) + } + return tx.Commit(ctx) +} diff --git a/services/protocols/internal/infrastructure/postgres/mail_store_read.go b/services/protocols/internal/infrastructure/postgres/mail_store_read.go new file mode 100644 index 0000000..7b53943 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/mail_store_read.go @@ -0,0 +1,202 @@ +package postgres + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +// The read side of MailStore. +// +// IMAP (LIST, SELECT, FETCH) and the webmail (folder list, message list) used to +// read the same tables with different queries, and so could answer differently: +// IMAP took a folder's message count from a cached column, the webmail counted +// the rows. A client comparing the two saw numbers that disagreed. They now ask +// the same question, and the answer is derived from the messages themselves, so +// a folder can only ever report what is actually in it. + +// FolderInfo is one folder with the numbers a client needs. +type FolderInfo struct { + ID string + Name string + SpecialUse string + UIDNext int64 + UIDValidity int64 + HighestModSeq int64 + // Total and Unread are counted from the messages, not read from the cached + // columns: the columns are kept true by every write path, and counting here + // makes a wrong one visible instead of repeated. + Total int + Unread int +} + +const folderColumns = ` + SELECT f.id::text, f.name, COALESCE(f.special_use, ''), + f.uid_next, f.uid_validity, f.highest_modseq, + COUNT(e.id)::int, + COUNT(e.id) FILTER ( + WHERE NOT EXISTS (SELECT 1 FROM message_flags mf + WHERE mf.message_id = e.id + AND mf.received_at = e.received_at + AND mf.flag = '\Seen') + )::int + FROM folders f + LEFT JOIN email_messages e ON e.folder_id = f.id AND e.expunged_at IS NULL` + +func scanFolder(row pgx.Row) (FolderInfo, error) { + var f FolderInfo + err := row.Scan(&f.ID, &f.Name, &f.SpecialUse, &f.UIDNext, &f.UIDValidity, &f.HighestModSeq, &f.Total, &f.Unread) + return f, err +} + +// Folders lists a mailbox's folders by name. +func (s *MailStore) Folders(ctx context.Context, mailboxID string) ([]FolderInfo, error) { + rows, err := s.pool.Query(ctx, folderColumns+` + WHERE f.mailbox_id = $1 + GROUP BY f.id + ORDER BY f.name`, mailboxID) + if err != nil { + return nil, fmt.Errorf("list folders: %w", err) + } + defer rows.Close() + + var out []FolderInfo + for rows.Next() { + f, err := scanFolder(rows) + if err != nil { + return nil, err + } + out = append(out, f) + } + return out, rows.Err() +} + +// FolderByName finds one folder by its exact name; nil, nil when there is none +// (IMAP SELECT treats that as a normal answer, not a failure). +func (s *MailStore) FolderByName(ctx context.Context, mailboxID, name string) (*FolderInfo, error) { + f, err := scanFolder(s.pool.QueryRow(ctx, folderColumns+` + WHERE f.mailbox_id = $1 AND f.name = $2 + GROUP BY f.id`, mailboxID, name)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + return &f, nil +} + +// MessageInfo is one message as either a mail client or the webmail lists it. +type MessageInfo struct { + UID uint32 + BlobID uuid.UUID + Size int64 + ReceivedAt time.Time + ModSeq uint64 + Flags []string + Subject string + SenderAddress string + FromName string + Snippet string + HasAttachments bool + SpamVerdict string + DmarcResult string +} + +// HasFlag reports whether the message carries the flag. +func (m MessageInfo) HasFlag(flag string) bool { + for _, f := range m.Flags { + if f == flag { + return true + } + } + return false +} + +// MessageQuery shapes a listing. The zero value lists every message of the +// folder by UID ascending, which is what IMAP needs to number them. +type MessageQuery struct { + // NewestFirst orders by arrival, newest first, as the webmail shows them + NewestFirst bool + // Search keeps messages whose subject, sender or snippet contains it + Search string + // Limit and Offset page the result; Limit 0 means no limit + Limit int + Offset int +} + +// Messages lists a folder's live messages with their flags. +func (s *MailStore) Messages(ctx context.Context, folderID string, q MessageQuery) ([]MessageInfo, error) { + order := `e.uid ASC` + if q.NewestFirst { + order = `e.received_at DESC, e.uid DESC` + } + limit := q.Limit + if limit <= 0 { + limit = 1<<31 - 1 + } + offset := q.Offset + if offset < 0 { + offset = 0 + } + + rows, err := s.pool.Query(ctx, ` + SELECT e.uid, e.blob_id, e.size_bytes, e.received_at, e.modseq, + COALESCE((SELECT array_agg(mf.flag) FROM message_flags mf + WHERE mf.message_id = e.id AND mf.received_at = e.received_at), '{}'), + COALESCE(e.subject, ''), e.sender_address::text, + COALESCE(e.from_display_name, ''), COALESCE(e.snippet, ''), + e.has_attachments, COALESCE(e.spam_verdict, ''), COALESCE(e.dmarc_result, '') + FROM email_messages e + WHERE e.folder_id = $1 AND e.expunged_at IS NULL + AND ($2 = '' OR e.subject ILIKE '%' || $2 || '%' + OR e.sender_address::text ILIKE '%' || $2 || '%' + OR e.snippet ILIKE '%' || $2 || '%') + ORDER BY `+order+` + LIMIT $3 OFFSET $4`, folderID, q.Search, limit, offset) + if err != nil { + return nil, fmt.Errorf("list messages: %w", err) + } + defer rows.Close() + + var out []MessageInfo + for rows.Next() { + var m MessageInfo + if err := rows.Scan(&m.UID, &m.BlobID, &m.Size, &m.ReceivedAt, &m.ModSeq, &m.Flags, + &m.Subject, &m.SenderAddress, &m.FromName, &m.Snippet, + &m.HasAttachments, &m.SpamVerdict, &m.DmarcResult); err != nil { + return nil, err + } + out = append(out, m) + } + return out, rows.Err() +} + +// MessageBlob resolves one message to its stored blob, or ErrMessageNotFound. +func (s *MailStore) MessageBlob(ctx context.Context, folderID string, uid uint32) (uuid.UUID, error) { + var blobID uuid.UUID + err := s.pool.QueryRow(ctx, ` + SELECT blob_id FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL`, + folderID, int64(uid)).Scan(&blobID) + if errors.Is(err, pgx.ErrNoRows) { + return uuid.Nil, ErrMessageNotFound + } + return blobID, err +} + +// MailboxIDByAddress resolves an active address to its mailbox id. +func (s *MailStore) MailboxIDByAddress(ctx context.Context, address string) (string, error) { + var id string + err := s.pool.QueryRow(ctx, ` + SELECT m.id::text FROM mailboxes m + JOIN domains d ON d.id = m.domain_id + WHERE m.email_address = $1 AND m.is_active AND d.is_active`, address).Scan(&id) + if errors.Is(err, pgx.ErrNoRows) { + return "", ErrMessageNotFound + } + return id, err +} diff --git a/services/protocols/internal/infrastructure/postgres/mailbox_state_migration_test.go b/services/protocols/internal/infrastructure/postgres/mailbox_state_migration_test.go new file mode 100644 index 0000000..81be194 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/mailbox_state_migration_test.go @@ -0,0 +1,158 @@ +package postgres + +import ( + "context" + "os" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" +) + +// runMigration0014 applies the migration file itself, so the test cannot drift +// from what production runs. It is idempotent by design, which is also tested. +func runMigration0014(t *testing.T, pool *pgxpool.Pool) { + t.Helper() + sql, err := os.ReadFile("../../../../../migrations/0014_mailbox_state_single_source.up.sql") + if err != nil { + t.Fatalf("read migration: %v", err) + } + if _, err := pool.Exec(context.Background(), string(sql)); err != nil { + t.Fatalf("apply migration 0014: %v", err) + } +} + +func folderNames(t *testing.T, pool *pgxpool.Pool, mailboxID string) map[string]string { + t.Helper() + rows, err := pool.Query(context.Background(), `SELECT name, COALESCE(special_use, '') FROM folders WHERE mailbox_id = $1`, mailboxID) + if err != nil { + t.Fatalf("list folders: %v", err) + } + defer rows.Close() + out := map[string]string{} + for rows.Next() { + var name, role string + if err := rows.Scan(&name, &role); err != nil { + t.Fatalf("scan: %v", err) + } + out[name] = role + } + return out +} + +// Numbers that two paths left wrong are corrected from the rows. +func TestMigration0014_ReconcilesDriftedCountersAndQuota(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + mailboxID := seedMailbox(t, pool) + deliver(t, pool, mailboxID, "INBOX", "a", false) + deliver(t, pool, mailboxID, "INBOX", "b", true) + deliver(t, pool, mailboxID, "Sent", "c", true) + + // The drift as it was found: counters and quota that disagree with the rows + if _, err := pool.Exec(ctx, `UPDATE folders SET unread_count = 9, total_count = 7 WHERE mailbox_id = $1`, mailboxID); err != nil { + t.Fatalf("corrupt counters: %v", err) + } + if _, err := pool.Exec(ctx, `UPDATE mailboxes SET used_bytes = 12345 WHERE id = $1`, mailboxID); err != nil { + t.Fatalf("corrupt quota: %v", err) + } + + runMigration0014(t, pool) + assertConsistent(t, pool, mailboxID, "the reconciliation") + + runMigration0014(t, pool) + assertConsistent(t, pool, mailboxID, "running the reconciliation a second time") +} + +// A mailbox that is already right is left exactly as it was. +func TestMigration0014_LeavesATrueMailboxAlone(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + mailboxID := seedMailbox(t, pool) + deliver(t, pool, mailboxID, "INBOX", "a", false) + before, usedBefore, _ := mailState(t, pool, mailboxID) + + runMigration0014(t, pool) + + after, usedAfter, _ := mailState(t, pool, mailboxID) + for name, f := range before { + if after[name].CachedTotal != f.CachedTotal || after[name].CachedUnread != f.CachedUnread || after[name].HighestModSeq != f.HighestModSeq { + t.Errorf("%s changed: %+v -> %+v", name, f, after[name]) + } + } + if usedAfter != usedBefore { + t.Errorf("used_bytes changed from %d to %d", usedBefore, usedAfter) + } + _ = ctx +} + +// The one list of standard folders: what is missing is added, what is there is +// not duplicated, and running it again changes nothing. +func TestProvisionDefaultFolders(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + runMigration0014(t, pool) + + domainID, mailboxID := uuid.New(), uuid.New() + name := "prov" + uuid.NewString()[:8] + ".test" + if _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name, is_active) VALUES ($1,$2,$3,true)`, domainID, name, name); err != nil { + t.Fatalf("seed domain: %v", err) + } + t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM domains WHERE id = $1`, domainID) }) + if _, err := pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash, is_active) VALUES ($1,$2,'u',$3,'x',true)`, + mailboxID, domainID, "u@"+name); err != nil { + t.Fatalf("seed mailbox: %v", err) + } + + want := map[string]string{"INBOX": "inbox", "Sent": "sent", "Drafts": "drafts", "Trash": "trash", "Junk": "junk", "Archive": "archive", "Reports": ""} + + if _, err := pool.Exec(ctx, `SELECT provision_default_folders($1)`, mailboxID); err != nil { + t.Fatalf("provision: %v", err) + } + got := folderNames(t, pool, mailboxID.String()) + for n, role := range want { + if r, ok := got[n]; !ok || r != role { + t.Errorf("folder %q: present=%v role=%q, want role %q", n, ok, r, role) + } + } + if len(got) != len(want) { + t.Errorf("got %d folders, want %d: %v", len(got), len(want), got) + } + + if _, err := pool.Exec(ctx, `SELECT provision_default_folders($1)`, mailboxID); err != nil { + t.Fatalf("provision again: %v", err) + } + if again := folderNames(t, pool, mailboxID.String()); len(again) != len(want) { + t.Errorf("a second call changed the folder count to %d", len(again)) + } +} + +// A folder that exists under another capitalisation counts as present, and the +// folder keeps the owner's mail. +func TestProvisionDefaultFolders_DoesNotDuplicateByCase(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + runMigration0014(t, pool) + mailboxID := seedMailbox(t, pool) // INBOX, Sent, Drafts, Trash + + if _, err := pool.Exec(ctx, `INSERT INTO folders (mailbox_id, name, special_use) VALUES ($1, 'reports', NULL)`, mailboxID); err != nil { + t.Fatalf("seed: %v", err) + } + if _, err := pool.Exec(ctx, `SELECT provision_default_folders($1)`, mailboxID); err != nil { + t.Fatalf("provision: %v", err) + } + + got := folderNames(t, pool, mailboxID) + reports := 0 + for n := range got { + if n == "reports" || n == "Reports" { + reports++ + } + } + if reports != 1 { + t.Errorf("found %d reports folders, want 1: %v", reports, got) + } + if len(got) != 7 { + t.Errorf("got %d folders, want 7: %v", len(got), got) + } +} diff --git a/services/protocols/internal/infrastructure/postgres/message_query_repository.go b/services/protocols/internal/infrastructure/postgres/message_query_repository.go index 09a0c0c..9f647ef 100644 --- a/services/protocols/internal/infrastructure/postgres/message_query_repository.go +++ b/services/protocols/internal/infrastructure/postgres/message_query_repository.go @@ -8,37 +8,26 @@ import ( "lambdamail/protocols/internal/application/port" ) -// MessageQueryRepository implements port.MessageQueryRepository against Postgres. +// MessageQueryRepository lists a folder's messages for IMAP FETCH, through the +// same MailStore listing the webmail uses. type MessageQueryRepository struct { - pool *pgxpool.Pool + store *MailStore } func NewMessageQueryRepository(pool *pgxpool.Pool) *MessageQueryRepository { - return &MessageQueryRepository{pool: pool} + return &MessageQueryRepository{store: NewMailStore(pool)} } func (r *MessageQueryRepository) ListMessages(ctx context.Context, folderID string) ([]port.MessageRecord, error) { - rows, err := r.pool.Query(ctx, ` - SELECT m.uid, m.blob_id, m.size_bytes, m.received_at, m.modseq, - COALESCE(array_agg(f.flag) FILTER (WHERE f.flag IS NOT NULL), '{}') - FROM email_messages m - LEFT JOIN message_flags f ON f.message_id = m.id AND f.received_at = m.received_at - WHERE m.folder_id = $1 AND m.expunged_at IS NULL - GROUP BY m.uid, m.blob_id, m.size_bytes, m.received_at, m.modseq - ORDER BY m.uid ASC - `, folderID) + messages, err := r.store.Messages(ctx, folderID, MessageQuery{}) if err != nil { return nil, err } - defer rows.Close() - var out []port.MessageRecord - for rows.Next() { - var rec port.MessageRecord - if err := rows.Scan(&rec.UID, &rec.BlobID, &rec.SizeBytes, &rec.ReceivedAt, &rec.ModSeq, &rec.Flags); err != nil { - return nil, err - } - out = append(out, rec) + for _, m := range messages { + out = append(out, port.MessageRecord{ + UID: m.UID, BlobID: m.BlobID, SizeBytes: m.Size, ReceivedAt: m.ReceivedAt, Flags: m.Flags, ModSeq: m.ModSeq, + }) } - return out, rows.Err() + return out, nil } diff --git a/services/protocols/internal/infrastructure/postgres/webmail_repository.go b/services/protocols/internal/infrastructure/postgres/webmail_repository.go index f39c8ca..bd1521f 100644 --- a/services/protocols/internal/infrastructure/postgres/webmail_repository.go +++ b/services/protocols/internal/infrastructure/postgres/webmail_repository.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sort" "github.com/google/uuid" "github.com/jackc/pgx/v5" @@ -19,456 +20,182 @@ var ErrMessageNotFound = errors.New("webmail: message not found") // WebmailRepository implements the webmail read side against Postgres. type WebmailRepository struct { - pool *pgxpool.Pool + pool *pgxpool.Pool + store *MailStore } func NewWebmailRepository(pool *pgxpool.Pool) *WebmailRepository { - return &WebmailRepository{pool: pool} + return &WebmailRepository{pool: pool, store: NewMailStore(pool)} } +// The methods below only read. They do not query on their own: the webmail asks +// MailStore the same questions IMAP does and shapes the answers for the +// browser, so a folder shows the same numbers in both. + func (r *WebmailRepository) FindMailboxIDByAddress(ctx context.Context, address string) (string, error) { - var id string - err := r.pool.QueryRow(ctx, ` - SELECT m.id::text FROM mailboxes m - JOIN domains d ON d.id = m.domain_id - WHERE m.email_address = $1 AND m.is_active AND d.is_active - `, address).Scan(&id) - if errors.Is(err, pgx.ErrNoRows) { - return "", ErrMessageNotFound - } - return id, err + return r.store.MailboxIDByAddress(ctx, address) } +// roleOrder is how the folder list is arranged for people: the folders they use +// every day first, in a fixed order, then their own by name. +var roleOrder = map[string]int{"inbox": 0, "drafts": 1, "sent": 2, "archive": 3, "junk": 4, "trash": 5} + func (r *WebmailRepository) ListFolders(ctx context.Context, mailboxID string) ([]port.WebmailFolder, error) { - // The counts are derived from the messages themselves rather than read out - // of folders.unread_count/total_count. - // - // Those columns are a denormalised cache maintained by every write path, - // and once any one of them was wrong nothing brought it back: the number - // the user saw stayed wrong through a reload, a logout and a fresh login, - // because the wrongness was stored. Deriving here is authoritative and - // self-healing - a folder can only ever report what is actually in it. - // - // The columns stay maintained for IMAP's sake (STATUS answers from them), - // so this is a correction at the read side, not their removal. - rows, err := r.pool.Query(ctx, ` - SELECT f.id::text, f.name, COALESCE(f.special_use, ''), - COUNT(e.id) FILTER ( - WHERE NOT EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id - AND mf.received_at = e.received_at - AND mf.flag = '\Seen') - )::int AS unread_count, - COUNT(e.id)::int AS total_count - FROM folders f - LEFT JOIN email_messages e - ON e.folder_id = f.id AND e.expunged_at IS NULL - WHERE f.mailbox_id = $1 - GROUP BY f.id, f.name, f.special_use - ORDER BY CASE f.special_use - WHEN 'inbox' THEN 0 WHEN 'drafts' THEN 1 WHEN 'sent' THEN 2 - WHEN 'archive' THEN 3 WHEN 'junk' THEN 4 WHEN 'trash' THEN 5 - ELSE 6 END, f.name - `, mailboxID) + folders, err := r.store.Folders(ctx, mailboxID) if err != nil { - return nil, fmt.Errorf("list folders: %w", err) + return nil, err } - defer rows.Close() - folders := []port.WebmailFolder{} - for rows.Next() { - var f port.WebmailFolder - if err := rows.Scan(&f.ID, &f.Name, &f.SpecialUse, &f.UnreadCount, &f.TotalCount); err != nil { - return nil, err + out := make([]port.WebmailFolder, 0, len(folders)) + for _, f := range folders { + out = append(out, port.WebmailFolder{ID: f.ID, Name: f.Name, SpecialUse: f.SpecialUse, UnreadCount: f.Unread, TotalCount: f.Total}) + } + + rank := func(f port.WebmailFolder) int { + if n, ok := roleOrder[f.SpecialUse]; ok { + return n } - folders = append(folders, f) + return 6 } - return folders, rows.Err() + sort.SliceStable(out, func(i, j int) bool { + if ri, rj := rank(out[i]), rank(out[j]); ri != rj { + return ri < rj + } + return out[i].Name < out[j].Name + }) + return out, nil } -// folderCondition matches either the special-use role ("inbox") or the literal -// folder name, so the UI can address a folder the same way IMAP does. -const folderCondition = `(f.special_use = LOWER($2) OR LOWER(f.name) = LOWER($2))` - func (r *WebmailRepository) ListMessages( ctx context.Context, mailboxID, folderName, search string, limit, offset int, ) ([]port.WebmailMessage, error) { if limit <= 0 || limit > 200 { limit = 50 } - if offset < 0 { - offset = 0 - } - rows, err := r.pool.Query(ctx, ` - SELECT e.uid, COALESCE(e.subject, ''), e.sender_address::text, - COALESCE(e.from_display_name, ''), COALESCE(e.snippet, ''), - e.size_bytes, e.received_at, e.has_attachments, - EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Seen'), - EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Flagged'), - EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Answered'), - COALESCE(e.spam_verdict, ''), COALESCE(e.dmarc_result, '') - FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.expunged_at IS NULL - AND ($3 = '' OR e.subject ILIKE '%%' || $3 || '%%' - OR e.sender_address::text ILIKE '%%' || $3 || '%%' - OR e.snippet ILIKE '%%' || $3 || '%%') - ORDER BY e.received_at DESC, e.uid DESC - LIMIT $4 OFFSET $5 - `, mailboxID, folderName, search, limit, offset) + folderID, err := r.store.ResolveFolder(ctx, mailboxID, folderName) + if errors.Is(err, port.ErrFolderMissing) { + return []port.WebmailMessage{}, nil + } if err != nil { - return nil, fmt.Errorf("list messages: %w", err) + return nil, err } - defer rows.Close() - messages := []port.WebmailMessage{} - for rows.Next() { - var m port.WebmailMessage - if err := rows.Scan( - &m.UID, &m.Subject, &m.SenderAddress, &m.FromName, &m.Snippet, - &m.SizeBytes, &m.ReceivedAt, &m.HasAttachments, - &m.Seen, &m.Flagged, &m.Answered, &m.SpamVerdict, &m.DmarcResult, - ); err != nil { - return nil, err - } - messages = append(messages, m) + messages, err := r.store.Messages(ctx, folderID, MessageQuery{NewestFirst: true, Search: search, Limit: limit, Offset: offset}) + if err != nil { + return nil, err } - return messages, rows.Err() + + out := make([]port.WebmailMessage, 0, len(messages)) + for _, m := range messages { + out = append(out, port.WebmailMessage{ + UID: m.UID, Subject: m.Subject, SenderAddress: m.SenderAddress, FromName: m.FromName, Snippet: m.Snippet, + SizeBytes: m.Size, ReceivedAt: m.ReceivedAt, HasAttachments: m.HasAttachments, + Seen: m.HasFlag(`\Seen`), Flagged: m.HasFlag(`\Flagged`), Answered: m.HasFlag(`\Answered`), + SpamVerdict: m.SpamVerdict, DmarcResult: m.DmarcResult, + }) + } + return out, nil } -func (r *WebmailRepository) GetMessageBlob( - ctx context.Context, mailboxID, folderName string, uid uint32, -) (uuid.UUID, error) { - var blobID uuid.UUID - err := r.pool.QueryRow(ctx, ` - SELECT e.blob_id FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.uid = $3 AND e.expunged_at IS NULL - `, mailboxID, folderName, int64(uid)).Scan(&blobID) - if errors.Is(err, pgx.ErrNoRows) { - return uuid.Nil, ErrMessageNotFound +func (r *WebmailRepository) GetMessageBlob(ctx context.Context, mailboxID, folderName string, uid uint32) (uuid.UUID, error) { + folderID, err := r.messageFolder(ctx, mailboxID, folderName) + if err != nil { + return uuid.Nil, err } - return blobID, err + return r.store.MessageBlob(ctx, folderID, uid) } -// MarkSeen keeps the \Seen flag and the folder's unread counter in step. They -// are updated together in one transaction, because a counter that disagrees -// with the flags is what makes an inbox show unread mail that is not there. -func (r *WebmailRepository) MarkSeen( - ctx context.Context, mailboxID, folderName string, uid uint32, seen bool, -) error { - tx, err := r.pool.Begin(ctx) +// The methods below change state. They do not do it themselves: the webmail +// names a message by mailbox, folder name and UID, MailStore by folder id and +// UID, and these translate. Every counter, modseq and quota rule lives in +// MailStore, which IMAP uses too, so the two can no longer disagree. + +// messageFolder resolves the folder a message is addressed in. A folder that is +// not there and a message that is not there look the same to the caller, so a +// stranger cannot probe which folders exist. +func (r *WebmailRepository) messageFolder(ctx context.Context, mailboxID, folderName string) (string, error) { + id, err := r.store.ResolveFolder(ctx, mailboxID, folderName) + if errors.Is(err, port.ErrFolderMissing) { + return "", ErrMessageNotFound + } + return id, err +} + +// MarkSeen sets or clears \Seen. The unread counter follows inside MailStore. +func (r *WebmailRepository) MarkSeen(ctx context.Context, mailboxID, folderName string, uid uint32, seen bool) error { + return r.setFlag(ctx, mailboxID, folderName, uid, `\Seen`, seen) +} + +// MarkFlagged sets or clears \Flagged. +func (r *WebmailRepository) MarkFlagged(ctx context.Context, mailboxID, folderName string, uid uint32, flagged bool) error { + return r.setFlag(ctx, mailboxID, folderName, uid, `\Flagged`, flagged) +} + +func (r *WebmailRepository) setFlag(ctx context.Context, mailboxID, folderName string, uid uint32, flag string, on bool) error { + folderID, err := r.messageFolder(ctx, mailboxID, folderName) if err != nil { return err } - defer tx.Rollback(ctx) + op := port.FlagOpAdd + if !on { + op = port.FlagOpDel + } + _, err = r.store.SetFlags(ctx, folderID, uid, op, []string{flag}, 0) + return err +} - var messageID uuid.UUID - var receivedAt any - var folderID string - err = tx.QueryRow(ctx, ` - SELECT e.id, e.received_at, f.id::text FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.uid = $3 AND e.expunged_at IS NULL - `, mailboxID, folderName, int64(uid)).Scan(&messageID, &receivedAt, &folderID) - if errors.Is(err, pgx.ErrNoRows) { +// Expunge removes one message for good (soft delete) and gives its bytes back. +func (r *WebmailRepository) Expunge(ctx context.Context, mailboxID, folderName string, uid uint32) error { + folderID, err := r.messageFolder(ctx, mailboxID, folderName) + if err != nil { + return err + } + removed, err := r.store.Expunge(ctx, folderID, []uint32{uid}) + if err != nil { + return err + } + if removed == 0 { return ErrMessageNotFound } - if err != nil { - return err - } - - var changed int64 - if seen { - tag, err := tx.Exec(ctx, ` - INSERT INTO message_flags (message_id, received_at, flag) - VALUES ($1, $2, '\Seen') ON CONFLICT DO NOTHING - `, messageID, receivedAt) - if err != nil { - return err - } - changed = tag.RowsAffected() - } else { - tag, err := tx.Exec(ctx, ` - DELETE FROM message_flags WHERE message_id = $1 AND received_at = $2 AND flag = '\Seen' - `, messageID, receivedAt) - if err != nil { - return err - } - changed = tag.RowsAffected() - } - - // Only a real transition moves the counter, so opening an already-read - // message repeatedly cannot drive unread_count negative. - if changed > 0 { - delta := 1 - if seen { - delta = -1 - } - if _, err := tx.Exec(ctx, ` - UPDATE folders SET unread_count = GREATEST(0, unread_count + $2), highest_modseq = highest_modseq + 1 - WHERE id = $1 - `, folderID, delta); err != nil { - return err - } - } - - return tx.Commit(ctx) -} - -// MarkFlagged sets or clears \Flagged. Unlike \Seen it moves no counter, but a -// real change still bumps the folder's modseq: that is how an IMAP client -// holding the folder open learns the flag changed without a full resync. -func (r *WebmailRepository) MarkFlagged( - ctx context.Context, mailboxID, folderName string, uid uint32, flagged bool, -) error { - tx, err := r.pool.Begin(ctx) - if err != nil { - return err - } - defer tx.Rollback(ctx) - - var messageID uuid.UUID - var receivedAt any - var folderID string - err = tx.QueryRow(ctx, ` - SELECT e.id, e.received_at, f.id::text FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.uid = $3 AND e.expunged_at IS NULL - `, mailboxID, folderName, int64(uid)).Scan(&messageID, &receivedAt, &folderID) - if errors.Is(err, pgx.ErrNoRows) { - return ErrMessageNotFound - } - if err != nil { - return err - } - - var changed int64 - if flagged { - tag, err := tx.Exec(ctx, ` - INSERT INTO message_flags (message_id, received_at, flag) - VALUES ($1, $2, '\Flagged') ON CONFLICT DO NOTHING - `, messageID, receivedAt) - if err != nil { - return err - } - changed = tag.RowsAffected() - } else { - tag, err := tx.Exec(ctx, ` - DELETE FROM message_flags WHERE message_id = $1 AND received_at = $2 AND flag = '\Flagged' - `, messageID, receivedAt) - if err != nil { - return err - } - changed = tag.RowsAffected() - } - - if changed > 0 { - if _, err := tx.Exec(ctx, - `UPDATE folders SET highest_modseq = highest_modseq + 1 WHERE id = $1`, folderID); err != nil { - return err - } - } - return tx.Commit(ctx) -} - -// Expunge marks one message deleted and keeps the folder counters in step. -// -// Soft delete, matching the expunged_at column the rest of the schema uses: the -// blob stays referenced until the retention sweep, so a mistaken delete is -// recoverable and IMAP clients holding the UID still resolve it. -// -// The message is located by mailbox as well as UID, so a caller cannot expunge -// somebody else's mail by guessing a number. -func (r *WebmailRepository) Expunge( - ctx context.Context, mailboxID, folderName string, uid uint32, -) error { - tx, err := r.pool.Begin(ctx) - if err != nil { - return err - } - defer tx.Rollback(ctx) - - var messageID uuid.UUID - var folderID string - var wasUnread bool - err = tx.QueryRow(ctx, ` - SELECT e.id, f.id::text, - NOT EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id - AND mf.received_at = e.received_at - AND mf.flag = '\Seen') - FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.uid = $3 AND e.expunged_at IS NULL - `, mailboxID, folderName, int64(uid)).Scan(&messageID, &folderID, &wasUnread) - if errors.Is(err, pgx.ErrNoRows) { - return ErrMessageNotFound - } - if err != nil { - return err - } - - if _, err := tx.Exec(ctx, - `UPDATE email_messages SET expunged_at = NOW() WHERE id = $1`, messageID); err != nil { - return err - } - - // GREATEST guards the counters against ever going negative, which a - // double expunge of the same row would otherwise cause. - unreadDelta := 0 - if wasUnread { - unreadDelta = -1 - } - if _, err := tx.Exec(ctx, ` - UPDATE folders - SET total_count = GREATEST(0, total_count - 1), - unread_count = GREATEST(0, unread_count + $2), - highest_modseq = highest_modseq + 1 - WHERE id = $1 - `, folderID, unreadDelta); err != nil { - return err - } - - return tx.Commit(ctx) -} - -// MoveToTrash relocates one message into the mailbox's Trash folder and -// returns the UID it was given there. -// -// This is what "delete" means in a mail client, and the webmail had no way to -// do it at all: the only removal in the codebase was Expunge, reachable solely -// when an autosave superseded a draft. A message the user wanted rid of - the -// empty draft left behind by a sent message, most visibly - could not be -// removed from any screen. -// -// The message keeps its identity and its blob; only the folder and the UID -// change, because a UID is meaningful only within one folder. Deleting from -// Trash itself is Expunge's job, not this one's. -func (r *WebmailRepository) MoveToTrash( - ctx context.Context, mailboxID, folderName string, uid uint32, -) (uint32, error) { - return r.moveInto(ctx, mailboxID, folderName, uid, trashTarget) -} - -// MoveToFolder files one message into a named folder. -// -// The target is matched on its special-use role or its name, the same way the -// rest of this repository addresses folders, so "archive" and "Archive" both -// reach the same place and a folder the user created themselves - which has no -// role - is reachable by name. -func (r *WebmailRepository) MoveToFolder( - ctx context.Context, mailboxID, folderName string, uid uint32, target string, -) (uint32, error) { - return r.moveInto(ctx, mailboxID, folderName, uid, target) + return nil } // trashTarget is what MoveToTrash resolves; kept as a constant so the two // callers cannot drift. const trashTarget = "trash" -func (r *WebmailRepository) moveInto( - ctx context.Context, mailboxID, folderName string, uid uint32, target string, -) (uint32, error) { - tx, err := r.pool.Begin(ctx) - if err != nil { - return 0, err - } - defer tx.Rollback(ctx) - - var messageID uuid.UUID - var sourceFolderID string - var wasUnread bool - err = tx.QueryRow(ctx, ` - SELECT e.id, f.id::text, - NOT EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id - AND mf.received_at = e.received_at - AND mf.flag = '\Seen') - FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.uid = $3 AND e.expunged_at IS NULL - `, mailboxID, folderName, int64(uid)).Scan(&messageID, &sourceFolderID, &wasUnread) - if errors.Is(err, pgx.ErrNoRows) { - return 0, ErrMessageNotFound - } - if err != nil { - return 0, err - } - - // Deleting something already in Trash has no further destination, so the - // caller is told to expunge instead of the message being moved onto itself. - var trashFolderID string - var trashUID int64 - err = tx.QueryRow(ctx, ` - SELECT id::text, uid_next FROM folders - WHERE mailbox_id = $1 AND (special_use = LOWER($2) OR LOWER(name) = LOWER($2)) - ORDER BY special_use IS NOT NULL DESC - LIMIT 1 - FOR UPDATE - `, mailboxID, target).Scan(&trashFolderID, &trashUID) - if errors.Is(err, pgx.ErrNoRows) { - return 0, port.ErrNoTrashFolder - } - if err != nil { - return 0, err - } - if trashFolderID == sourceFolderID { - return 0, port.ErrAlreadyInTrash - } - - var modseq int64 - if err := tx.QueryRow(ctx, ` - UPDATE folders SET uid_next = uid_next + 1, highest_modseq = highest_modseq + 1, - total_count = total_count + 1, unread_count = unread_count + $2 - WHERE id = $1 - RETURNING highest_modseq - `, trashFolderID, boolToInt(wasUnread)).Scan(&modseq); err != nil { - return 0, err - } - - if _, err := tx.Exec(ctx, ` - UPDATE email_messages SET folder_id = $2, uid = $3, modseq = $4 - WHERE id = $1 - `, messageID, trashFolderID, trashUID, modseq); err != nil { - return 0, err - } - - unreadDelta := 0 - if wasUnread { - unreadDelta = -1 - } - if _, err := tx.Exec(ctx, ` - UPDATE folders - SET total_count = GREATEST(0, total_count - 1), - unread_count = GREATEST(0, unread_count + $2), - highest_modseq = highest_modseq + 1 - WHERE id = $1 - `, sourceFolderID, unreadDelta); err != nil { - return 0, err - } - - if err := tx.Commit(ctx); err != nil { - return 0, err - } - return uint32(trashUID), nil +// MoveToTrash is what "delete" means in a mail client: the message goes to the +// Trash folder and keeps its identity. Deleting from Trash itself is Expunge's +// job, so asking to move out of Trash reports port.ErrAlreadyInTrash. +func (r *WebmailRepository) MoveToTrash(ctx context.Context, mailboxID, folderName string, uid uint32) (uint32, error) { + return r.moveInto(ctx, mailboxID, folderName, uid, trashTarget) } -func boolToInt(b bool) int { - if b { - return 1 +// MoveToFolder files one message into a named folder, matched by role or name. +func (r *WebmailRepository) MoveToFolder(ctx context.Context, mailboxID, folderName string, uid uint32, target string) (uint32, error) { + return r.moveInto(ctx, mailboxID, folderName, uid, target) +} + +func (r *WebmailRepository) moveInto(ctx context.Context, mailboxID, folderName string, uid uint32, target string) (uint32, error) { + sourceID, err := r.messageFolder(ctx, mailboxID, folderName) + if err != nil { + return 0, err } - return 0 + destID, err := r.store.ResolveFolder(ctx, mailboxID, target) + if errors.Is(err, port.ErrFolderMissing) { + if target == trashTarget { + return 0, port.ErrNoTrashFolder + } + return 0, port.ErrFolderMissing + } + if err != nil { + return 0, err + } + if destID == sourceID { + return 0, port.ErrAlreadyInTrash + } + return r.store.Move(ctx, sourceID, uid, destID) } // --- folders the mailbox owner creates for themselves -------------------- @@ -507,20 +234,12 @@ func (r *WebmailRepository) RenameFolder(ctx context.Context, mailboxID, from, t return nil } -// DeleteFolder removes a folder and the mail filed in it. -// -// The messages are expunged rather than deleted outright, matching what the -// rest of the schema does: the blobs stay referenced until the retention sweep, -// so a folder deleted by mistake is still recoverable from the database. +// DeleteFolder removes a folder the owner made, and the mail filed in it. The +// messages are expunged (recoverable) and their bytes go back to the quota, +// both inside MailStore. func (r *WebmailRepository) DeleteFolder(ctx context.Context, mailboxID, name string) error { - tx, err := r.pool.Begin(ctx) - if err != nil { - return err - } - defer tx.Rollback(ctx) - var folderID string - err = tx.QueryRow(ctx, ` + err := r.pool.QueryRow(ctx, ` SELECT id::text FROM folders WHERE mailbox_id = $1 AND LOWER(name) = LOWER($2) AND special_use IS NULL `, mailboxID, name).Scan(&folderID) @@ -530,16 +249,5 @@ func (r *WebmailRepository) DeleteFolder(ctx context.Context, mailboxID, name st if err != nil { return err } - - if _, err := tx.Exec(ctx, ` - UPDATE email_messages SET expunged_at = NOW() - WHERE folder_id = $1 AND expunged_at IS NULL - `, folderID); err != nil { - return fmt.Errorf("expunge the folder's messages: %w", err) - } - - if _, err := tx.Exec(ctx, `DELETE FROM folders WHERE id = $1`, folderID); err != nil { - return fmt.Errorf("delete folder %q: %w", name, err) - } - return tx.Commit(ctx) + return r.store.DeleteFolder(ctx, folderID) }