LambdaMail/services/protocols/internal/infrastructure/postgres/report_ingestion_test.go

package postgres

import (
	"bytes"
	"compress/gzip"
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"io"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"

	"lambdamail/protocols/internal/application/port"
	"lambdamail/protocols/internal/application/usecase"
	"lambdamail/protocols/internal/domain/entity"
	"lambdamail/protocols/internal/infrastructure/diskstorage"
)

// End to end over a real database: a report arrives as mail, is parsed into
// the report tables, and is filed into Reports without raising an unread badge.

// The report Google actually sends, with the domain changed.
const liveTlsRptReport = `{"organization-name":"Google Inc.",` +
	`"date-range":{"start-datetime":"2026-08-19T00:00:00Z","end-datetime":"2026-08-19T23:59:59Z"},` +
	`"contact-info":"smtp-tls-reporting@google.com","report-id":"2026-08-19T00:00:00Z_example.test",` +
	`"policies":[{"policy":{"policy-type":"no-policy-found","policy-domain":"example.test"},` +
	`"summary":{"total-successful-session-count":2,"total-failure-session-count":0}}]}`

func reportMail(t *testing.T, filename string, content []byte) []byte {
	t.Helper()
	var gzBuf bytes.Buffer
	gz := gzip.NewWriter(&gzBuf)
	if _, err := gz.Write(content); err != nil {
		t.Fatal(err)
	}
	if err := gz.Close(); err != nil {
		t.Fatal(err)
	}
	encoded := base64.StdEncoding.EncodeToString(gzBuf.Bytes())

	var b bytes.Buffer
	b.WriteString("From: smtp-tls-reporting@google.com\r\n")
	b.WriteString("Subject: Report Domain: example.test\r\n")
	b.WriteString("MIME-Version: 1.0\r\n")
	b.WriteString("Content-Type: multipart/mixed; boundary=\"sep\"\r\n\r\n")
	b.WriteString("--sep\r\n")
	b.WriteString("Content-Type: text/plain; charset=utf-8\r\n\r\nReport attached.\r\n")
	b.WriteString("--sep\r\n")
	b.WriteString("Content-Type: application/gzip; name=\"" + filename + "\"\r\n")
	b.WriteString("Content-Disposition: attachment; filename=\"" + filename + "\"\r\n")
	b.WriteString("Content-Transfer-Encoding: base64\r\n\r\n")
	b.WriteString(encoded + "\r\n")
	b.WriteString("--sep--\r\n")
	return b.Bytes()
}

func TestDeliveredTlsRptReportIsIngestedAndFiled(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	mailboxID := seedMailbox(t, pool)

	// The Reports folder migration 0008 adds for every mailbox.
	if _, err := pool.Exec(ctx,
		`INSERT INTO folders (mailbox_id, name, special_use) VALUES ($1, 'Reports', NULL)`,
		uuid.MustParse(mailboxID)); err != nil {
		t.Fatalf("seed Reports folder: %v", err)
	}

	reportRepo := NewReportRepository(pool)
	inbound := usecase.NewProcessInboundEmailUseCase(
		nil, &dbBlobStore{pool: pool, dir: t.TempDir()}, NewInboundMessageRepository(pool))
	inbound.SetReportIngestor(
		usecase.NewIngestDeliveredReportsUseCase(usecase.NewIngestReportsUseCase(reportRepo)))

	reportID := "delivered-" + uuid.NewString()
	payload := reportMail(t,
		"google.com!example.test!1787097600!1787183999!001.json.gz",
		[]byte(strings.Replace(liveTlsRptReport,
			"2026-08-19T00:00:00Z_example.test", reportID, 1)))
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(),
			`DELETE FROM tls_rpt_reports WHERE report_id = $1`, reportID)
	})

	if err := inbound.Handle(ctx, usecase.ProcessInboundEmailInput{
		Sender:             "smtp-tls-reporting@google.com",
		Recipients:         []port.MailboxRecord{{ID: uuid.MustParse(mailboxID)}},
		RecipientAddresses: []string{"tlsrpt@example.test"},
		Body:               bytes.NewReader(payload),
	}); err != nil {
		t.Fatalf("Handle: %v", err)
	}

	// The report reached the table the admin console reads.
	var stored int
	if err := pool.QueryRow(ctx,
		`SELECT COUNT(*) FROM tls_rpt_reports WHERE report_id = $1`, reportID).Scan(&stored); err != nil {
		t.Fatalf("count stored reports: %v", err)
	}
	if stored != 1 {
		t.Errorf("the report was not stored: %d rows", stored)
	}

	// And the message went to Reports, silently.
	webmail := NewWebmailRepository(pool)
	folders, err := webmail.ListFolders(ctx, mailboxID)
	if err != nil {
		t.Fatalf("list folders: %v", err)
	}
	for _, f := range folders {
		switch f.Name {
		case "Reports":
			if f.TotalCount != 1 {
				t.Errorf("Reports holds %d messages, want 1", f.TotalCount)
			}
			if f.UnreadCount != 0 {
				t.Errorf("Reports shows %d unread; a machine-read report raises no badge", f.UnreadCount)
			}
		case "INBOX":
			if f.TotalCount != 0 {
				t.Errorf("the report landed in the inbox: %d messages", f.TotalCount)
			}
		}
	}
}

// dbBlobStore writes a real message_blobs row so the delivery's foreign keys
// hold, without needing the disk storage driver in a database test.
type dbBlobStore struct {
	pool *pgxpool.Pool
	// dir is where the bytes actually land, so the real disk reader can read
	// them back. Pointing storage_path at a file that does not exist would
	// leave the read path untested.
	dir string
}

func (s *dbBlobStore) Store(ctx context.Context, r io.Reader) (port.BlobRef, error) {
	body, err := io.ReadAll(r)
	if err != nil {
		return port.BlobRef{}, err
	}
	sum := sha256.Sum256(body)
	digest := hex.EncodeToString(sum[:])
	// Deduplicated on the digest, as the real storage drivers are: identical
	// bytes are one blob, and a second delivery of the same report reuses it.
	id := uuid.New()
	path := filepath.Join(s.dir, id.String()+".eml")
	if err := os.WriteFile(path, body, 0o600); err != nil {
		return port.BlobRef{}, err
	}
	if err := s.pool.QueryRow(ctx, `
		INSERT INTO message_blobs (id, content_sha256, storage_driver, storage_path, size_bytes, ref_count)
		VALUES ($1, $2, 'local', $3, $4, 0)
		ON CONFLICT (content_sha256) DO UPDATE SET storage_path = EXCLUDED.storage_path
		RETURNING id`,
		id, digest, path, len(body)).Scan(&id); err != nil {
		return port.BlobRef{}, err
	}
	return port.BlobRef{ID: id, SHA256: digest, SizeBytes: int64(len(body))}, nil
}

// A reporter that retries - or an SMTP retry after a slow commit - delivers
// the same report twice. Automatic ingestion makes that routine, where before
// reports were only ever written by hand through the API, so the same report
// arriving again must not double every number the console shows.
func TestIngestingTheSameReportTwiceStoresItOnce(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	repo := NewReportRepository(pool)

	reportID := "retry-" + uuid.NewString()
	report := func() *entity.TlsRptReport {
		return &entity.TlsRptReport{
			OrganizationName: "Google Inc.",
			ReportID:         reportID,
			Domain:           "example.test",
			DateRangeBegin:   time.Now().Add(-24 * time.Hour),
			DateRangeEnd:     time.Now(),
			Policies: []entity.TlsRptPolicy{
				{PolicyType: "no-policy-found", SuccessCount: 2, FailureCount: 0},
			},
		}
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(),
			`DELETE FROM tls_rpt_reports WHERE report_id = $1`, reportID)
	})

	if err := repo.SaveTlsRptReport(ctx, report()); err != nil {
		t.Fatalf("first save: %v", err)
	}
	if err := repo.SaveTlsRptReport(ctx, report()); err != nil {
		t.Fatalf("a redelivered report must not error: %v", err)
	}

	var headers, policies int
	if err := pool.QueryRow(ctx,
		`SELECT COUNT(*) FROM tls_rpt_reports WHERE report_id = $1`, reportID).Scan(&headers); err != nil {
		t.Fatal(err)
	}
	if headers != 1 {
		t.Errorf("the same report is stored %d times", headers)
	}
	if err := pool.QueryRow(ctx, `
		SELECT COUNT(*) FROM tls_rpt_report_policies p
		  JOIN tls_rpt_reports r ON r.id = p.report_id
		 WHERE r.report_id = $1`, reportID).Scan(&policies); err != nil {
		t.Fatal(err)
	}
	if policies != 1 {
		t.Errorf("the report's policies are stored %d times", policies)
	}
}

// The same, for DMARC: its unique key includes domain_id, which this path
// never fills, and in Postgres a NULL never conflicts - so the key it had did
// not in fact prevent a duplicate.
func TestIngestingTheSameDmarcReportTwiceStoresItOnce(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	repo := NewReportRepository(pool)

	reportID := "retry-" + uuid.NewString()
	report := func() *entity.DmarcReport {
		return &entity.DmarcReport{
			OrgName:        "google.com",
			ReportID:       reportID,
			Domain:         "example.test",
			DateRangeBegin: time.Now().Add(-24 * time.Hour),
			DateRangeEnd:   time.Now(),
			Records: []entity.DmarcRecord{
				{SourceIP: "203.0.113.10", Count: 3, Disposition: "none", DKIMResult: "pass", SPFResult: "pass"},
			},
		}
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(),
			`DELETE FROM dmarc_reports WHERE report_id = $1`, reportID)
	})

	if err := repo.SaveDmarcReport(ctx, report()); err != nil {
		t.Fatalf("first save: %v", err)
	}
	if err := repo.SaveDmarcReport(ctx, report()); err != nil {
		t.Fatalf("a redelivered report must not error: %v", err)
	}

	var headers int
	if err := pool.QueryRow(ctx,
		`SELECT COUNT(*) FROM dmarc_reports WHERE report_id = $1`, reportID).Scan(&headers); err != nil {
		t.Fatal(err)
	}
	if headers != 1 {
		t.Errorf("the same DMARC report is stored %d times", headers)
	}
}

// The backfill, over a real database: a report that was delivered to the
// inbox before ingestion existed is parsed after the fact and filed away.
func TestBackfillParsesAReportAlreadySittingInTheInbox(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	mailboxID := seedMailbox(t, pool)

	if _, err := pool.Exec(ctx,
		`INSERT INTO folders (mailbox_id, name, special_use) VALUES ($1, 'Reports', NULL)`,
		uuid.MustParse(mailboxID)); err != nil {
		t.Fatalf("seed Reports folder: %v", err)
	}

	// Delivered the old way: straight into the inbox, unread, never parsed.
	reportID := "backfill-" + uuid.NewString()
	body := `{"organization-name":"Google Inc.",` +
		`"date-range":{"start-datetime":"2026-08-19T00:00:00Z","end-datetime":"2026-08-19T23:59:59Z"},` +
		`"report-id":"` + reportID + `",` +
		`"policies":[{"policy":{"policy-type":"sts","policy-domain":"example.test"},` +
		`"summary":{"total-successful-session-count":5,"total-failure-session-count":1}}]}`
	payload := reportMail(t, "google.com!example.test!1!2!001.json.gz", []byte(body))

	blobs := &dbBlobStore{pool: pool, dir: t.TempDir()}
	blob, err := blobs.Store(ctx, bytes.NewReader(payload))
	if err != nil {
		t.Fatalf("store blob: %v", err)
	}
	uids, err := NewInboundMessageRepository(pool).PersistAll(ctx, []port.PersistInboundMessageInput{{
		MailboxID:        uuid.MustParse(mailboxID),
		Blob:             blob,
		SenderAddress:    "smtp-tls-reporting@google.com",
		RecipientAddress: "tlsrpt@example.test",
		TargetFolderName: "INBOX",
		Subject:          "Report Domain: example.test",
	}})
	if err != nil || len(uids) != 1 {
		t.Fatalf("seed delivered report: %v", err)
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(),
			`DELETE FROM tls_rpt_reports WHERE report_id = $1`, reportID)
	})

	store := NewDeliveredReportStore(pool)
	uc := usecase.NewBackfillReportsUseCase(store, diskstorage.NewLocalDiskBlobReader(pool),
		usecase.NewIngestDeliveredReportsUseCase(
			usecase.NewIngestReportsUseCase(NewReportRepository(pool))))

	summary, err := uc.Run(ctx, 0)
	if err != nil {
		t.Fatalf("backfill: %v", err)
	}
	if summary.Ingested < 1 {
		t.Fatalf("the backlogged report was not ingested: %+v", summary)
	}

	var stored int
	if err := pool.QueryRow(ctx,
		`SELECT COUNT(*) FROM tls_rpt_reports WHERE report_id = $1`, reportID).Scan(&stored); err != nil {
		t.Fatal(err)
	}
	if stored != 1 {
		t.Errorf("the report reached the table %d times, want 1", stored)
	}

	// And it left the inbox.
	webmail := NewWebmailRepository(pool)
	folders, _ := webmail.ListFolders(ctx, mailboxID)
	for _, f := range folders {
		if f.Name == "INBOX" && f.TotalCount != 0 {
			t.Errorf("the backfilled report is still in the inbox: %d messages", f.TotalCount)
		}
		if f.Name == "Reports" {
			if f.TotalCount != 1 {
				t.Errorf("Reports holds %d messages, want 1", f.TotalCount)
			}
			if f.UnreadCount != 0 {
				t.Errorf("the backfilled report raised an unread badge: %d", f.UnreadCount)
			}
		}
	}

	// Running it again finds nothing left to do.
	second, err := uc.Run(ctx, 0)
	if err != nil {
		t.Fatalf("second run: %v", err)
	}
	if second.Ingested != 0 {
		t.Errorf("a second pass re-ingested %d reports; it should find none", second.Ingested)
	}
}