#4: implement F2 sub-project 1 - IMAP core (auth, LIST, SELECT, FETCH, STORE)
Adds a real :143 IMAP listener to the Go protocols service: a mailbox owner authenticates over IMAP (Argon2id against mailboxes.password_hash, enumeration-resistant per PLAN.md section 14.6), lists folders, selects INBOX, fetches messages F1's SMTP path already persisted - including real ENVELOPE/BODY[] built from the actual stored MIME bytes via go-imap/v2's own ExtractEnvelope/ExtractBodySection helpers, not reconstructed from denormalized columns - and sets/reads flags via STORE. - ImapSessionUseCase orchestrates auth, folder listing/selection, message querying, and flag mutation against new Postgres repositories (auth, folders, message query, flags) and a local-disk blob reader. - A github.com/emersion/go-imap/v2 (pinned v2.0.0-beta.8) Backend/Session implements 7 real methods (Login, List, Select, Unselect, Status, Fetch, Store) plus a correct no-op Poll; the other 10 of the interface's 18 methods (Create/Delete/Rename/Subscribe/Unsubscribe/Append/Idle/Expunge/ Search/Copy) are explicit stubs returning a clear "not yet implemented" IMAP error - SEARCH/EXPUNGE/COPY/MOVE, IDLE/CONDSTORE, mailbox management, and APPEND are separate follow-up sub-projects per the design doc's agreed ordering with the user. Verified end-to-end with a real go-imap/v2 imapclient (embedded Postgres, no Docker needed) driving Login -> Select -> Fetch -> Store -> List against a message seeded through F1's real SMTP use case, proving true SMTP-write/IMAP-read interop, not just isolated unit tests. Built via 8 sequential tasks under subagent-driven development, each independently reviewed, plus a final whole-branch review. Two real bugs were found and fixed before merge, both exactly the kind unit tests with fakes cannot catch: - Store's UID-set resolution silently no-op'd on dynamic ranges like "UID STORE 1:*" (the common "mark all as read" operation) after an earlier draft bypassed the folder's real message list; restored to resolve against real messages via the same pattern Fetch already used correctly, with tests proving both dynamic ranges and non-UID STORE. - go-imap/v2's server calls Session.Poll() implicitly before the status response of nearly every authenticated command (FETCH/STORE/SEARCH/ APPEND/COPY), not just IDLE as assumed - a hard-stubbed Poll broke every real command through this path; fixed to a correct no-op. - List had zero test coverage despite being committed scope; added unit tests for wildcard/delimiter matching plus a real E2E LIST assertion. Known follow-up, not blocking: a pre-existing test-fixture hygiene gap (hardcoded content_sha256 blob hashes with no cleanup, shared across several Postgres integration test files including one new to this branch) causes local-dev test collisions on repeated runs against a persistent database - invisible to CI (fresh Postgres per run) but worth a cleanup pass.
This commit is contained in:
Lucas Rayan committed
1 parent
fc8bbaa3ce
commit
9fa19291d4
25 files changed
+1920
-1
No files matched your search
@@ -0,0 +1,42 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"lambdamail/protocols/internal/application/port"
|
||||
)
|
||||
|
||||
// AuthRepository implements port.AuthRepository against Postgres.
|
||||
type AuthRepository struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewAuthRepository(pool *pgxpool.Pool) *AuthRepository {
|
||||
return &AuthRepository{pool: pool}
|
||||
}
|
||||
|
||||
func (r *AuthRepository) FindByAddress(ctx context.Context, address string) (*port.MailboxAuth, error) {
|
||||
row := r.pool.QueryRow(ctx, `
|
||||
SELECT m.id, m.password_hash
|
||||
FROM mailboxes m
|
||||
JOIN domains d ON d.id = m.domain_id
|
||||
WHERE m.email_address = $1
|
||||
AND m.is_active = true
|
||||
AND d.is_active = true
|
||||
`, address)
|
||||
|
||||
var rec port.MailboxAuth
|
||||
var id uuid.UUID
|
||||
if err := row.Scan(&id, &rec.PasswordHash); err != nil {
|
||||
if err == pgx.ErrNoRows {
|
||||
return nil, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
rec.ID = id
|
||||
return &rec, nil
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
func TestAuthRepository_FindByAddress_ReturnsHashForActiveMailbox(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
defer pool.Close()
|
||||
ctx := context.Background()
|
||||
|
||||
domainID := uuid.New()
|
||||
domainName := "authrepo-test-" + domainID.String() + ".invalid"
|
||||
_, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`,
|
||||
domainID, domainName, domainName)
|
||||
if err != nil {
|
||||
t.Fatalf("seed domain: %v", err)
|
||||
}
|
||||
defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID)
|
||||
|
||||
mailboxID := uuid.New()
|
||||
address := "user@" + domainName
|
||||
hash := "$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG"
|
||||
_, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', $3, $4)`,
|
||||
mailboxID, domainID, address, hash)
|
||||
if err != nil {
|
||||
t.Fatalf("seed mailbox: %v", err)
|
||||
}
|
||||
|
||||
repo := NewAuthRepository(pool)
|
||||
rec, err := repo.FindByAddress(ctx, address)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if rec == nil {
|
||||
t.Fatal("rec = nil, want a record")
|
||||
}
|
||||
if rec.ID != mailboxID || rec.PasswordHash != hash {
|
||||
t.Errorf("rec = %+v, want ID=%v PasswordHash=%q", rec, mailboxID, hash)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthRepository_FindByAddress_ReturnsNilForUnknownAddress(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
defer pool.Close()
|
||||
|
||||
repo := NewAuthRepository(pool)
|
||||
rec, err := repo.FindByAddress(context.Background(), "definitely-not-seeded@example.invalid")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if rec != nil {
|
||||
t.Errorf("rec = %+v, want nil", rec)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"lambdamail/protocols/internal/application/port"
|
||||
)
|
||||
|
||||
// FlagRepository implements port.FlagRepository against Postgres.
|
||||
type FlagRepository struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewFlagRepository(pool *pgxpool.Pool) *FlagRepository {
|
||||
return &FlagRepository{pool: pool}
|
||||
}
|
||||
|
||||
func (r *FlagRepository) SetFlags(ctx context.Context, folderID string, uid uint32, op port.FlagOp, flags []string) error {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("begin transaction: %w", err)
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
var messageID string
|
||||
var receivedAt any
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT id, received_at FROM email_messages WHERE folder_id = $1 AND uid = $2
|
||||
`, folderID, uid).Scan(&messageID, &receivedAt); err != nil {
|
||||
return fmt.Errorf("find message folder=%s uid=%d: %w", folderID, uid, 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
|
||||
`, messageID, receivedAt, flag); err != nil {
|
||||
return 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
|
||||
`, messageID, receivedAt, flag); err != nil {
|
||||
return 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`, messageID, receivedAt); err != nil {
|
||||
return 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
|
||||
`, messageID, receivedAt, flag); err != nil {
|
||||
return fmt.Errorf("set flag %q: %w", flag, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return fmt.Errorf("commit: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sort"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"lambdamail/protocols/internal/application/port"
|
||||
)
|
||||
|
||||
func TestFlagRepository_SetFlags_AddThenDelThenSet(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
defer pool.Close()
|
||||
ctx := context.Background()
|
||||
|
||||
domainID := uuid.New()
|
||||
domainName := "flagrepo-test-" + domainID.String() + ".invalid"
|
||||
_, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`,
|
||||
domainID, domainName, domainName)
|
||||
if err != nil {
|
||||
t.Fatalf("seed domain: %v", err)
|
||||
}
|
||||
defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID)
|
||||
|
||||
mailboxID := uuid.New()
|
||||
_, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG')`,
|
||||
mailboxID, domainID, "user@"+domainName)
|
||||
if err != nil {
|
||||
t.Fatalf("seed mailbox: %v", err)
|
||||
}
|
||||
folderID := uuid.New()
|
||||
_, err = pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'INBOX', 'inbox')`,
|
||||
folderID, mailboxID)
|
||||
if err != nil {
|
||||
t.Fatalf("seed folder: %v", err)
|
||||
}
|
||||
blobID := uuid.New()
|
||||
_, err = pool.Exec(ctx, `INSERT INTO message_blobs (id, content_sha256, storage_driver, storage_path, size_bytes) VALUES ($1, $2, 'local', '/tmp/x', 10)`,
|
||||
blobID, "1111110000000000000000000000000000000000000000000000000000000000"[:64])
|
||||
if err != nil {
|
||||
t.Fatalf("seed blob: %v", err)
|
||||
}
|
||||
messageID := uuid.New()
|
||||
_, err = pool.Exec(ctx, `
|
||||
INSERT INTO email_messages (id, mailbox_id, folder_id, uid, blob_id, sender_address, recipient_addresses, size_bytes)
|
||||
VALUES ($1, $2, $3, 7, $4, 'sender@example.test', $5, 10)
|
||||
`, messageID, mailboxID, folderID, blobID, []string{"user@" + domainName})
|
||||
if err != nil {
|
||||
t.Fatalf("seed message: %v", err)
|
||||
}
|
||||
|
||||
readFlags := func() []string {
|
||||
t.Helper()
|
||||
rows, err := pool.Query(ctx, `SELECT flag FROM message_flags WHERE message_id = $1 ORDER BY flag`, messageID)
|
||||
if err != nil {
|
||||
t.Fatalf("query flags: %v", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var flags []string
|
||||
for rows.Next() {
|
||||
var f string
|
||||
if err := rows.Scan(&f); err != nil {
|
||||
t.Fatalf("scan flag: %v", err)
|
||||
}
|
||||
flags = append(flags, f)
|
||||
}
|
||||
return flags
|
||||
}
|
||||
|
||||
repo := NewFlagRepository(pool)
|
||||
|
||||
if err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpAdd, []string{"\\Seen", "\\Flagged"}); err != nil {
|
||||
t.Fatalf("add: %v", err)
|
||||
}
|
||||
got := readFlags()
|
||||
sort.Strings(got)
|
||||
if len(got) != 2 || got[0] != "\\Flagged" || got[1] != "\\Seen" {
|
||||
t.Fatalf("after add: flags = %v, want [\\Flagged \\Seen]", got)
|
||||
}
|
||||
|
||||
if err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpDel, []string{"\\Flagged"}); err != nil {
|
||||
t.Fatalf("del: %v", err)
|
||||
}
|
||||
got = readFlags()
|
||||
if len(got) != 1 || got[0] != "\\Seen" {
|
||||
t.Fatalf("after del: flags = %v, want [\\Seen]", got)
|
||||
}
|
||||
|
||||
if err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpSet, []string{"\\Answered"}); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
got = readFlags()
|
||||
if len(got) != 1 || got[0] != "\\Answered" {
|
||||
t.Fatalf("after set: flags = %v, want [\\Answered] (Set replaces the full flag set)", got)
|
||||
}
|
||||
|
||||
// A client sending "STORE 1 FLAGS (\Seen \Seen)" - a duplicate flag in a
|
||||
// single command - must not fail the whole STORE with a primary key
|
||||
// violation on the second INSERT for the same (message_id, received_at,
|
||||
// flag).
|
||||
if err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpSet, []string{"\\Seen", "\\Seen"}); err != nil {
|
||||
t.Fatalf("set with duplicate flags in input: %v", err)
|
||||
}
|
||||
got = readFlags()
|
||||
if len(got) != 1 || got[0] != "\\Seen" {
|
||||
t.Fatalf("after set with duplicate input flags: flags = %v, want [\\Seen]", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
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.
|
||||
type ImapFolderRepository struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewImapFolderRepository(pool *pgxpool.Pool) *ImapFolderRepository {
|
||||
return &ImapFolderRepository{pool: pool}
|
||||
}
|
||||
|
||||
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
|
||||
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); err != nil {
|
||||
if err == pgx.ErrNoRows {
|
||||
return nil, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
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
|
||||
FROM folders f
|
||||
WHERE f.mailbox_id = $1
|
||||
ORDER BY f.name
|
||||
`, 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); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, rec)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
func TestImapFolderRepository_FindByName_ReturnsFolderWithUIDData(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
defer pool.Close()
|
||||
ctx := context.Background()
|
||||
|
||||
domainID := uuid.New()
|
||||
domainName := "imapfolder-test-" + domainID.String() + ".invalid"
|
||||
_, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`,
|
||||
domainID, domainName, domainName)
|
||||
if err != nil {
|
||||
t.Fatalf("seed domain: %v", err)
|
||||
}
|
||||
defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID)
|
||||
|
||||
mailboxID := uuid.New()
|
||||
_, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG')`,
|
||||
mailboxID, domainID, "user@"+domainName)
|
||||
if err != nil {
|
||||
t.Fatalf("seed mailbox: %v", err)
|
||||
}
|
||||
|
||||
folderID := uuid.New()
|
||||
_, err = pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use, uid_next, uid_validity) VALUES ($1, $2, 'INBOX', 'inbox', 5, 12345)`,
|
||||
folderID, mailboxID)
|
||||
if err != nil {
|
||||
t.Fatalf("seed folder: %v", err)
|
||||
}
|
||||
|
||||
repo := NewImapFolderRepository(pool)
|
||||
rec, err := repo.FindByName(ctx, mailboxID.String(), "INBOX")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if rec == nil {
|
||||
t.Fatal("rec = nil, want a folder record")
|
||||
}
|
||||
if rec.UIDNext != 5 || rec.UIDValidity != 12345 {
|
||||
t.Errorf("rec = %+v, want UIDNext=5 UIDValidity=12345", rec)
|
||||
}
|
||||
}
|
||||
|
||||
func TestImapFolderRepository_FindByName_ReturnsNilForUnknownFolder(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
defer pool.Close()
|
||||
|
||||
repo := NewImapFolderRepository(pool)
|
||||
rec, err := repo.FindByName(context.Background(), uuid.New().String(), "Nonexistent")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if rec != nil {
|
||||
t.Errorf("rec = %+v, want nil", rec)
|
||||
}
|
||||
}
|
||||
|
||||
func TestImapFolderRepository_ListFolders_ReturnsAllFoldersForMailbox(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
defer pool.Close()
|
||||
ctx := context.Background()
|
||||
|
||||
domainID := uuid.New()
|
||||
domainName := "imapfolder-list-" + domainID.String() + ".invalid"
|
||||
_, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`,
|
||||
domainID, domainName, domainName)
|
||||
if err != nil {
|
||||
t.Fatalf("seed domain: %v", err)
|
||||
}
|
||||
defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID)
|
||||
|
||||
mailboxID := uuid.New()
|
||||
_, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG')`,
|
||||
mailboxID, domainID, "user@"+domainName)
|
||||
if err != nil {
|
||||
t.Fatalf("seed mailbox: %v", err)
|
||||
}
|
||||
_, err = pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'INBOX', 'inbox')`,
|
||||
uuid.New(), mailboxID)
|
||||
if err != nil {
|
||||
t.Fatalf("seed INBOX: %v", err)
|
||||
}
|
||||
_, err = pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name) VALUES ($1, $2, 'Archive')`,
|
||||
uuid.New(), mailboxID)
|
||||
if err != nil {
|
||||
t.Fatalf("seed Archive: %v", err)
|
||||
}
|
||||
|
||||
repo := NewImapFolderRepository(pool)
|
||||
recs, err := repo.ListFolders(ctx, mailboxID.String())
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if len(recs) != 2 {
|
||||
t.Errorf("len(recs) = %d, want 2", len(recs))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"lambdamail/protocols/internal/application/port"
|
||||
)
|
||||
|
||||
// MessageQueryRepository implements port.MessageQueryRepository against Postgres.
|
||||
type MessageQueryRepository struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewMessageQueryRepository(pool *pgxpool.Pool) *MessageQueryRepository {
|
||||
return &MessageQueryRepository{pool: 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,
|
||||
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
|
||||
ORDER BY m.uid ASC
|
||||
`, folderID)
|
||||
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.Flags); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, rec)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
func TestMessageQueryRepository_ListMessages_ReturnsRowsOrderedByUID(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
defer pool.Close()
|
||||
ctx := context.Background()
|
||||
|
||||
domainID := uuid.New()
|
||||
domainName := "msgquery-test-" + domainID.String() + ".invalid"
|
||||
_, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`,
|
||||
domainID, domainName, domainName)
|
||||
if err != nil {
|
||||
t.Fatalf("seed domain: %v", err)
|
||||
}
|
||||
defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID)
|
||||
|
||||
mailboxID := uuid.New()
|
||||
_, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG')`,
|
||||
mailboxID, domainID, "user@"+domainName)
|
||||
if err != nil {
|
||||
t.Fatalf("seed mailbox: %v", err)
|
||||
}
|
||||
folderID := uuid.New()
|
||||
_, err = pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'INBOX', 'inbox')`,
|
||||
folderID, mailboxID)
|
||||
if err != nil {
|
||||
t.Fatalf("seed folder: %v", err)
|
||||
}
|
||||
blobID := uuid.New()
|
||||
_, err = pool.Exec(ctx, `INSERT INTO message_blobs (id, content_sha256, storage_driver, storage_path, size_bytes) VALUES ($1, $2, 'local', '/tmp/x', 42)`,
|
||||
blobID, "abcdef0000000000000000000000000000000000000000000000000000000000"[:64])
|
||||
if err != nil {
|
||||
t.Fatalf("seed blob: %v", err)
|
||||
}
|
||||
|
||||
for _, uidVal := range []int64{2, 1} { // insert out of order to prove the query sorts
|
||||
_, err = pool.Exec(ctx, `
|
||||
INSERT INTO email_messages (mailbox_id, folder_id, uid, blob_id, sender_address, recipient_addresses, size_bytes)
|
||||
VALUES ($1, $2, $3, $4, 'sender@example.test', $5, 42)
|
||||
`, mailboxID, folderID, uidVal, blobID, []string{"user@" + domainName})
|
||||
if err != nil {
|
||||
t.Fatalf("seed message uid=%d: %v", uidVal, err)
|
||||
}
|
||||
}
|
||||
|
||||
repo := NewMessageQueryRepository(pool)
|
||||
recs, err := repo.ListMessages(ctx, folderID.String())
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if len(recs) != 2 {
|
||||
t.Fatalf("len(recs) = %d, want 2", len(recs))
|
||||
}
|
||||
if recs[0].UID != 1 || recs[1].UID != 2 {
|
||||
t.Errorf("UIDs = [%d, %d], want [1, 2] (ascending order)", recs[0].UID, recs[1].UID)
|
||||
}
|
||||
if recs[0].BlobID != blobID {
|
||||
t.Errorf("BlobID = %v, want %v", recs[0].BlobID, blobID)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user