ultisuite-backend/internal/integrationtest/migration/delta_test.go

//go:build integration

package migration_test

import (
	"context"
	"net/http"
	"strings"
	"testing"

	"github.com/ultisuite/ulti-backend/internal/integrationtest"
	migr "github.com/ultisuite/ulti-backend/internal/migration"
	"github.com/ultisuite/ulti-backend/internal/users"
)

func TestMigrationCutoverResetsCompletedJobs(t *testing.T) {
	h := integrationtest.RequireHarness(t)
	ctx := context.Background()

	adminClient, adminClaims := integrationtest.RequireAdminClient(t, h)
	if _, err := users.EnsureUser(ctx, h.Pool, adminClaims); err != nil {
		t.Fatalf("ensure admin: %v", err)
	}
	if err := users.GrantPlatformAdmin(ctx, h.Pool, adminClaims.Sub); err != nil {
		t.Fatalf("grant admin: %v", err)
	}

	createResp, err := adminClient.Post("/api/v1/admin/migration/projects", map[string]any{
		"name":            "Cutover test",
		"source_provider": "google",
	})
	integrationtest.FailIf(err, t, "create project")
	integrationtest.FailUnlessStatus(t, createResp, 201)

	var created struct {
		ID string `json:"id"`
	}
	integrationtest.DecodeJSON(t, createResp, &created)

	actResp, err := adminClient.Post("/api/v1/admin/migration/projects/"+created.ID+"/activate", nil)
	integrationtest.FailIf(err, t, "activate project")
	integrationtest.FailUnlessStatus(t, actResp, 200)

	migrateeEmail := "cutover-" + created.ID[:8] + "@example.com"
	inviteResp, err := adminClient.Post("/api/v1/admin/migration/projects/"+created.ID+"/invites", map[string]string{
		"email": migrateeEmail,
	})
	integrationtest.FailIf(err, t, "create invite")
	integrationtest.FailUnlessStatus(t, inviteResp, 201)

	var invite struct {
		Token string `json:"token"`
	}
	integrationtest.DecodeJSON(t, inviteResp, &invite)

	migrateeClaims := integrationtest.RegularUser(integrationtest.NewExternalID("cutover"))
	migrateeClaims.Email = migrateeEmail
	migrateeClient, err := h.Client(migrateeClaims)
	integrationtest.FailIf(err, t, "migratee client")

	userID, err := users.EnsureUser(ctx, h.Pool, migrateeClaims)
	integrationtest.FailIf(err, t, "ensure migratee")

	claimResp, err := migrateeClient.Post("/api/v1/migration/claim", map[string]string{
		"token":    invite.Token,
		"password": "test-password-123",
	})
	integrationtest.FailIf(err, t, "claim invite")
	integrationtest.FailUnlessStatus(t, claimResp, 200)

	_, err = h.Pool.Exec(ctx, `
		UPDATE migration_jobs SET status = 'completed', updated_at = NOW()
		WHERE project_id = $1::uuid AND user_id = $2::uuid
	`, created.ID, userID)
	integrationtest.FailIf(err, t, "mark jobs completed")

	cutoverResp, err := adminClient.Post("/api/v1/admin/migration/projects/"+created.ID+"/cutover", nil)
	integrationtest.FailIf(err, t, "cutover")
	integrationtest.FailUnlessStatus(t, cutoverResp, 200)

	var cutover struct {
		Project struct {
			Status    string  `json:"status"`
			DeltaMode bool    `json:"delta_mode"`
			CutoverAt *string `json:"cutover_at"`
		} `json:"project"`
	}
	integrationtest.DecodeJSON(t, cutoverResp, &cutover)
	project := cutover.Project
	if project.Status != "cutover" || !project.DeltaMode || project.CutoverAt == nil {
		t.Fatalf("cutover project: %#v", project)
	}

	var pendingCount int
	if err := h.Pool.QueryRow(ctx, `
		SELECT COUNT(*) FROM migration_jobs
		WHERE project_id = $1::uuid AND user_id = $2::uuid AND status = 'pending'
	`, created.ID, userID).Scan(&pendingCount); err != nil {
		t.Fatalf("count pending jobs: %v", err)
	}
	if pendingCount != 4 {
		t.Fatalf("pending jobs = %d, want 4", pendingCount)
	}

	listResp, err := adminClient.Get("/api/v1/admin/migration/projects")
	integrationtest.FailIf(err, t, "list projects after cutover")
	integrationtest.FailUnlessStatus(t, listResp, 200)

	var listed struct {
		Projects []struct {
			ID         string `json:"id"`
			CutoverDNS *struct {
				Warnings []string `json:"warnings"`
			} `json:"cutover_dns"`
		} `json:"projects"`
	}
	integrationtest.DecodeJSON(t, listResp, &listed)
	var found bool
	for _, p := range listed.Projects {
		if p.ID != created.ID {
			continue
		}
		found = true
		if p.CutoverDNS == nil {
			t.Fatal("expected cutover_dns on listed project")
		}
	}
	if !found {
		t.Fatal("cutover project not found in list")
	}
}

func TestGoogleContactsDeltaDeletesRemoved(t *testing.T) {
	h := integrationtest.RequireHarness(t)
	ctx := context.Background()

	userID, email := insertMigrationTestUser(t, h.Pool, "contacts-delta")
	nc, _ := mockNextcloudClient(t, h.Pool, email)

	client := googleRewriteClient(t, func(w http.ResponseWriter, r *http.Request) {
		if strings.Contains(r.URL.Path, "/v1/people/me/connections") {
			_, _ = w.Write([]byte(`{
				"connections":[{
					"resourceName":"people/deleted-1",
					"metadata":{"deleted":true}
				}],
				"nextSyncToken":"sync-next"
			}`))
			return
		}
		http.NotFound(w, r)
	})

	importer := migr.NewContactsImporter(h.Pool, nc).WithHTTPClient(client)
	job := &migr.Job{
		UserID: userID,
		CursorJSON: map[string]any{
			"syncToken": "sync-old",
			"imported_ids": map[string]any{
				"people/deleted-1": true,
			},
		},
		StatsJSON: map[string]any{},
	}

	err := importer.ImportBatch(ctx, job, "token", "google", true, func(status string, cursor, stats map[string]any, jobErr string) error {
		if jobErr != "" {
			t.Fatalf("import error: %s", jobErr)
		}
		if status != "completed" {
			t.Fatalf("status = %q, want completed", status)
		}
		deleted, _ := stats["delta_deleted"].(float64)
		if deleted != 1 {
			t.Fatalf("delta_deleted = %v, want 1", stats["delta_deleted"])
		}
		if cursor["syncToken"] != "sync-next" {
			t.Fatalf("sync token = %v", cursor["syncToken"])
		}
		return nil
	})
	integrationtest.FailIf(err, t, "import batch")
}

func TestGoogleCalendarDeltaUpdatesExisting(t *testing.T) {
	h := integrationtest.RequireHarness(t)
	ctx := context.Background()

	userID, email := insertMigrationTestUser(t, h.Pool, "calendar-delta-update")
	nc, _ := mockNextcloudClient(t, h.Pool, email)

	client := googleRewriteClient(t, func(w http.ResponseWriter, r *http.Request) {
		switch {
		case strings.Contains(r.URL.Path, "/calendar/v3/users/me/calendarList"):
			_, _ = w.Write([]byte(`{"items":[{"id":"primary","summary":"Primary"}]}`))
		case strings.Contains(r.URL.Path, "/calendar/v3/calendars/") && strings.Contains(r.URL.Path, "/events"):
			_, _ = w.Write([]byte(`{
				"items":[{"id":"evt-1","status":"confirmed","summary":"Updated meeting","start":{"dateTime":"2026-06-13T10:00:00Z"},"end":{"dateTime":"2026-06-13T11:00:00Z"}}],
				"nextSyncToken":"cal-sync-next"
			}`))
		default:
			http.NotFound(w, r)
		}
	})

	importer := migr.NewCalendarImporter(h.Pool, nc).WithHTTPClient(client)
	job := &migr.Job{
		UserID: userID,
		CursorJSON: map[string]any{
			"calendarSyncTokens": map[string]any{"primary": "cal-sync-old"},
			"imported_ids":       map[string]any{"primary:evt-1": true},
		},
		StatsJSON: map[string]any{},
	}

	err := importer.ImportBatch(ctx, job, "token", "google", true, func(status string, cursor, stats map[string]any, jobErr string) error {
		if jobErr != "" {
			t.Fatalf("import error: %s", jobErr)
		}
		updated, _ := stats["delta_updated"].(float64)
		if updated != 1 {
			t.Fatalf("delta_updated = %v, want 1", stats["delta_updated"])
		}
		imported, _ := stats["delta_imported"].(float64)
		if imported != 0 {
			t.Fatalf("delta_imported = %v, want 0", stats["delta_imported"])
		}
		return nil
	})
	integrationtest.FailIf(err, t, "import batch")
}

func TestGoogleContactsDeltaUpdatesExisting(t *testing.T) {
	h := integrationtest.RequireHarness(t)
	ctx := context.Background()

	userID, email := insertMigrationTestUser(t, h.Pool, "contacts-delta-update")
	nc, _ := mockNextcloudClient(t, h.Pool, email)

	client := googleRewriteClient(t, func(w http.ResponseWriter, r *http.Request) {
		if strings.Contains(r.URL.Path, "/v1/people/me/connections") {
			_, _ = w.Write([]byte(`{
				"connections":[{
					"resourceName":"people/abc",
					"names":[{"displayName":"Alice Updated"}],
					"emailAddresses":[{"value":"alice@example.com"}]
				}],
				"nextSyncToken":"sync-next"
			}`))
			return
		}
		http.NotFound(w, r)
	})

	importer := migr.NewContactsImporter(h.Pool, nc).WithHTTPClient(client)
	job := &migr.Job{
		UserID: userID,
		CursorJSON: map[string]any{
			"syncToken": "sync-old",
			"imported_ids": map[string]any{
				"people/abc": true,
			},
		},
		StatsJSON: map[string]any{},
	}

	err := importer.ImportBatch(ctx, job, "token", "google", true, func(status string, cursor, stats map[string]any, jobErr string) error {
		if jobErr != "" {
			t.Fatalf("import error: %s", jobErr)
		}
		updated, _ := stats["delta_updated"].(float64)
		if updated != 1 {
			t.Fatalf("delta_updated = %v, want 1", stats["delta_updated"])
		}
		imported, _ := stats["delta_imported"].(float64)
		if imported != 0 {
			t.Fatalf("delta_imported = %v, want 0", stats["delta_imported"])
		}
		return nil
	})
	integrationtest.FailIf(err, t, "import batch")
}

func TestGoogleDriveDeltaDeletesRemovedFile(t *testing.T) {
	h := integrationtest.RequireHarness(t)
	ctx := context.Background()

	userID, email := insertMigrationTestUser(t, h.Pool, "drive-delta")
	nc, _ := mockNextcloudClient(t, h.Pool, email)

	client := googleRewriteClient(t, func(w http.ResponseWriter, r *http.Request) {
		if strings.Contains(r.URL.Path, "/drive/v3/changes") {
			_, _ = w.Write([]byte(`{
				"changes":[{"fileId":"file-removed","removed":true}],
				"newStartPageToken":"token-next"
			}`))
			return
		}
		http.NotFound(w, r)
	})

	importer := migr.NewDriveImporter(h.Pool, nc).WithHTTPClient(client)
	job := &migr.Job{
		UserID: userID,
		CursorJSON: map[string]any{
			"driveChangeToken": "token-old",
			"imported_paths": map[string]any{
				"file-removed": "Docs/report.docx",
			},
			"imported_ids": map[string]any{
				"file-removed": true,
			},
		},
		StatsJSON: map[string]any{},
	}

	err := importer.ImportBatch(ctx, job, "token", "google", true, func(status string, cursor, stats map[string]any, jobErr string) error {
		if jobErr != "" {
			t.Fatalf("import error: %s", jobErr)
		}
		if status != "completed" {
			t.Fatalf("status = %q, want completed", status)
		}
		deleted, _ := stats["delta_deleted"].(float64)
		if deleted != 1 {
			t.Fatalf("delta_deleted = %v, want 1", stats["delta_deleted"])
		}
		if cursor["driveChangeToken"] != "token-next" {
			t.Fatalf("change token = %v", cursor["driveChangeToken"])
		}
		if _, ok := cursor["imported_ids"]; ok {
			t.Fatal("expected imported_ids stripped from cursor")
		}
		if _, ok := cursor["imported_paths"]; ok {
			t.Fatal("expected imported_paths stripped from cursor")
		}
		return nil
	})
	integrationtest.FailIf(err, t, "import batch")
}

func TestGoogleCalendarDeltaDeletesCancelled(t *testing.T) {
	h := integrationtest.RequireHarness(t)
	ctx := context.Background()

	userID, email := insertMigrationTestUser(t, h.Pool, "calendar-delta")
	nc, _ := mockNextcloudClient(t, h.Pool, email)

	client := googleRewriteClient(t, func(w http.ResponseWriter, r *http.Request) {
		switch {
		case strings.Contains(r.URL.Path, "/calendar/v3/users/me/calendarList"):
			_, _ = w.Write([]byte(`{"items":[{"id":"primary","summary":"Primary"}]}`))
		case strings.Contains(r.URL.Path, "/calendar/v3/calendars/") && strings.Contains(r.URL.Path, "/events"):
			_, _ = w.Write([]byte(`{
				"items":[{"id":"evt-1","status":"cancelled","summary":"Old meeting"}],
				"nextSyncToken":"cal-sync-next"
			}`))
		default:
			http.NotFound(w, r)
		}
	})

	importer := migr.NewCalendarImporter(h.Pool, nc).WithHTTPClient(client)
	job := &migr.Job{
		UserID: userID,
		CursorJSON: map[string]any{
			"calendarSyncTokens": map[string]any{"primary": "cal-sync-old"},
			"imported_ids":       map[string]any{"primary:evt-1": true},
		},
		StatsJSON: map[string]any{},
	}

	err := importer.ImportBatch(ctx, job, "token", "google", true, func(status string, cursor, stats map[string]any, jobErr string) error {
		if jobErr != "" {
			t.Fatalf("import error: %s", jobErr)
		}
		deleted, _ := stats["delta_deleted"].(float64)
		if deleted != 1 {
			t.Fatalf("delta_deleted = %v, want 1", stats["delta_deleted"])
		}
		return nil
	})
	integrationtest.FailIf(err, t, "import batch")
}

func TestMicrosoftContactsDeltaRemoved(t *testing.T) {
	h := integrationtest.RequireHarness(t)
	ctx := context.Background()

	userID, email := insertMigrationTestUser(t, h.Pool, "ms-contacts-delta")
	nc, _ := mockNextcloudClient(t, h.Pool, email)

	client := graphRewriteClient(t, func(w http.ResponseWriter, r *http.Request) {
		if strings.Contains(r.URL.Path, "/v1.0/me/contacts") {
			_, _ = w.Write([]byte(`{
				"value":[{"id":"c-1","@removed":{"reason":"deleted"}}],
				"@odata.deltaLink":"https://graph.microsoft.com/v1.0/me/contacts/delta?token=next"
			}`))
			return
		}
		http.NotFound(w, r)
	})

	importer := migr.NewContactsImporter(h.Pool, nc).WithHTTPClient(client)
	job := &migr.Job{
		UserID: userID,
		CursorJSON: map[string]any{
			"deltaLink": "https://graph.microsoft.com/v1.0/me/contacts/delta?token=old",
			"imported_ids": map[string]any{
				"c-1": true,
			},
		},
		StatsJSON: map[string]any{},
	}

	err := importer.ImportBatch(ctx, job, "token", "microsoft", true, func(status string, cursor, stats map[string]any, jobErr string) error {
		if jobErr != "" {
			t.Fatalf("import error: %s", jobErr)
		}
		deleted, _ := stats["delta_deleted"].(float64)
		if deleted != 1 {
			t.Fatalf("delta_deleted = %v, want 1", stats["delta_deleted"])
		}
		return nil
	})
	integrationtest.FailIf(err, t, "import batch")
}

func TestGraphMailDeltaDeletesRemoved(t *testing.T) {
	h := integrationtest.RequireHarness(t)
	ctx := context.Background()

	userID, err := users.EnsureUser(ctx, h.Pool, integrationtest.RegularUser(integrationtest.NewExternalID("graph-delta-mail")))
	integrationtest.FailIf(err, t, "ensure user")

	var accountID string
	err = h.Pool.QueryRow(ctx, `
		INSERT INTO mail_accounts (user_id, email, provider, is_active)
		VALUES ($1::uuid, 'graph-delta@test.local', 'hosted', true)
		RETURNING id::text
	`, userID).Scan(&accountID)
	integrationtest.FailIf(err, t, "insert mail account")

	uid := migr.RemoteMessageUIDForTest("msg-removed-1")
	_, err = h.Pool.Exec(ctx, `
		INSERT INTO messages (account_id, folder_id, uid, message_id, subject, from_addr, to_addrs, date, snippet, body_text, body_html, flags, labels)
		SELECT $1::uuid, f.id, $2, '<test@local>', 'To delete', '[]', '[]', NOW(), '', '', '', '{}', '{}'
		FROM mail_folders f WHERE f.account_id = $1::uuid AND f.remote_name = 'INBOX' LIMIT 1
	`, accountID, uid)
	integrationtest.FailIf(err, t, "seed message")

	client := graphRewriteClient(t, func(w http.ResponseWriter, r *http.Request) {
		if strings.Contains(r.URL.Path, "/mailFolders") {
			_, _ = w.Write([]byte(`{"value":[{"id":"inbox-id","displayName":"Inbox","wellKnownName":"inbox"}]}`))
			return
		}
		if strings.Contains(r.URL.Path, "/messages") {
			_, _ = w.Write([]byte(`{
				"value":[{"id":"msg-removed-1","@removed":{"reason":"deleted"}}],
				"@odata.deltaLink":"https://graph.microsoft.com/v1.0/me/messages/delta?token=next"
			}`))
			return
		}
		http.NotFound(w, r)
	})

	importer := migr.NewGraphImporter(h.Pool).WithHTTPClient(client).WithBaseURL("https://graph.microsoft.com")
	job := &migr.Job{
		UserID:     userID,
		CursorJSON: map[string]any{"deltaLink": "https://graph.microsoft.com/v1.0/me/messages/delta?token=old"},
		StatsJSON:  map[string]any{},
	}

	err = importer.ImportBatch(ctx, job, "token", true, func(status string, cursor, stats map[string]any, jobErr string) error {
		if jobErr != "" {
			t.Fatalf("import error: %s", jobErr)
		}
		if status != "completed" {
			t.Fatalf("status = %q, want completed", status)
		}
		deleted, _ := stats["delta_deleted"].(float64)
		if deleted != 1 {
			t.Fatalf("delta_deleted = %v, want 1", stats["delta_deleted"])
		}
		return nil
	})
	integrationtest.FailIf(err, t, "import batch")

	var count int
	if err := h.Pool.QueryRow(ctx, `SELECT COUNT(*) FROM messages WHERE account_id = $1::uuid AND uid = $2`, accountID, uid).Scan(&count); err != nil {
		t.Fatalf("count messages: %v", err)
	}
	if count != 0 {
		t.Fatalf("message count = %d, want 0", count)
	}
}

func TestGraphFolderDeltaDeletesRemoved(t *testing.T) {
	h := integrationtest.RequireHarness(t)
	ctx := context.Background()

	userID, err := users.EnsureUser(ctx, h.Pool, integrationtest.RegularUser(integrationtest.NewExternalID("graph-folder-delta")))
	integrationtest.FailIf(err, t, "ensure user")

	var accountID string
	err = h.Pool.QueryRow(ctx, `
		INSERT INTO mail_accounts (user_id, email, provider, is_active)
		VALUES ($1::uuid, 'graph-folder-delta@test.local', 'hosted', true)
		RETURNING id::text
	`, userID).Scan(&accountID)
	integrationtest.FailIf(err, t, "insert mail account")

	uid := migr.RemoteMessageUIDForTest("msg-folder-removed")
	_, err = h.Pool.Exec(ctx, `
		INSERT INTO messages (account_id, folder_id, uid, message_id, subject, from_addr, to_addrs, date, snippet, body_text, body_html, flags, labels)
		SELECT $1::uuid, f.id, $2, '<test@local>', 'To delete', '[]', '[]', NOW(), '', '', '', '{}', '{}'
		FROM mail_folders f WHERE f.account_id = $1::uuid AND f.remote_name = 'INBOX' LIMIT 1
	`, accountID, uid)
	integrationtest.FailIf(err, t, "seed message")

	inboxID := "inbox-folder-id"
	sentID := "sent-folder-id"
	client := graphRewriteClient(t, func(w http.ResponseWriter, r *http.Request) {
		if strings.Contains(r.URL.Path, "/childFolders") {
			_, _ = w.Write([]byte(`{"value":[]}`))
			return
		}
		if strings.HasSuffix(r.URL.Path, "/mailFolders") {
			_, _ = w.Write([]byte(`{"value":[
				{"id":"` + inboxID + `","displayName":"Inbox","wellKnownName":"inbox"},
				{"id":"` + sentID + `","displayName":"Sent","wellKnownName":"sentitems"}
			]}`))
			return
		}
		if strings.Contains(r.URL.Path, "/mailFolders/"+inboxID+"/messages/delta") {
			_, _ = w.Write([]byte(`{
				"value":[{"id":"msg-folder-removed","@removed":{"reason":"deleted"}}],
				"@odata.deltaLink":"https://graph.microsoft.com/v1.0/me/mailFolders/` + inboxID + `/messages/delta?token=inbox-done"
			}`))
			return
		}
		if strings.Contains(r.URL.Path, "/mailFolders/"+sentID+"/messages/delta") {
			_, _ = w.Write([]byte(`{
				"value":[],
				"@odata.deltaLink":"https://graph.microsoft.com/v1.0/me/mailFolders/` + sentID + `/messages/delta?token=sent-done"
			}`))
			return
		}
		http.NotFound(w, r)
	})

	importer := migr.NewGraphImporter(h.Pool).WithHTTPClient(client).WithBaseURL("https://graph.microsoft.com")
	job := &migr.Job{
		UserID: userID,
		CursorJSON: map[string]any{
			"graphFolderQueue": []any{inboxID, sentID},
			"folderDeltaLinks": map[string]any{
				inboxID: "https://graph.microsoft.com/v1.0/me/mailFolders/" + inboxID + "/messages/delta?token=inbox-old",
				sentID:  "https://graph.microsoft.com/v1.0/me/mailFolders/" + sentID + "/messages/delta?token=sent-old",
			},
		},
		StatsJSON: map[string]any{},
	}

	for {
		var finalStatus string
		err = importer.ImportBatch(ctx, job, "token", true, func(status string, cursor, stats map[string]any, jobErr string) error {
			if jobErr != "" {
				t.Fatalf("import error: %s", jobErr)
			}
			finalStatus = status
			return nil
		})
		integrationtest.FailIf(err, t, "import batch")
		if finalStatus == "completed" {
			break
		}
	}

	deleted, _ := job.StatsJSON["delta_deleted"].(float64)
	if deleted != 1 {
		t.Fatalf("delta_deleted = %v, want 1", job.StatsJSON["delta_deleted"])
	}

	var count int
	if err := h.Pool.QueryRow(ctx, `SELECT COUNT(*) FROM messages WHERE account_id = $1::uuid AND uid = $2`, accountID, uid).Scan(&count); err != nil {
		t.Fatalf("count messages: %v", err)
	}
	if count != 0 {
		t.Fatalf("message count = %d, want 0", count)
	}
}

func TestGmailHistoryDeltaDeletesMessage(t *testing.T) {
	h := integrationtest.RequireHarness(t)
	ctx := context.Background()

	userID, err := users.EnsureUser(ctx, h.Pool, integrationtest.RegularUser(integrationtest.NewExternalID("gmail-delta-mail")))
	integrationtest.FailIf(err, t, "ensure user")

	var accountID string
	err = h.Pool.QueryRow(ctx, `
		INSERT INTO mail_accounts (user_id, email, provider, is_active)
		VALUES ($1::uuid, 'gmail-delta@test.local', 'hosted', true)
		RETURNING id::text
	`, userID).Scan(&accountID)
	integrationtest.FailIf(err, t, "insert mail account")

	gmailID := "abc123deleted"
	uid := migr.GmailUIDForTest(gmailID)
	_, err = h.Pool.Exec(ctx, `
		INSERT INTO messages (account_id, folder_id, uid, message_id, subject, from_addr, to_addrs, date, snippet, body_text, body_html, flags, labels)
		SELECT $1::uuid, f.id, $2, '<test@local>', 'To delete', '[]', '[]', NOW(), '', '', '', '{}', '{}'
		FROM mail_folders f WHERE f.account_id = $1::uuid AND f.remote_name = 'INBOX' LIMIT 1
	`, accountID, uid)
	integrationtest.FailIf(err, t, "seed message")

	client := googleRewriteClient(t, func(w http.ResponseWriter, r *http.Request) {
		if strings.Contains(r.URL.Path, "/gmail/v1/users/me/history") {
			_, _ = w.Write([]byte(`{
				"history":[{"messagesDeleted":[{"message":{"id":"` + gmailID + `"}}]}],
				"historyId":"99999"
			}`))
			return
		}
		http.NotFound(w, r)
	})

	importer := migr.NewGmailImporter(h.Pool).WithHTTPClient(client)
	job := &migr.Job{
		UserID:     userID,
		CursorJSON: map[string]any{"historyId": "88888"},
		StatsJSON:  map[string]any{},
	}

	err = importer.ImportBatch(ctx, job, "token", true, func(status string, cursor, stats map[string]any, jobErr string) error {
		if jobErr != "" {
			t.Fatalf("import error: %s", jobErr)
		}
		deleted, _ := stats["delta_deleted"].(float64)
		if deleted != 1 {
			t.Fatalf("delta_deleted = %v, want 1", stats["delta_deleted"])
		}
		if cursor["historyId"] != "99999" {
			t.Fatalf("historyId = %v", cursor["historyId"])
		}
		return nil
	})
	integrationtest.FailIf(err, t, "import batch")

	var count int
	if err := h.Pool.QueryRow(ctx, `SELECT COUNT(*) FROM messages WHERE account_id = $1::uuid AND uid = $2`, accountID, uid).Scan(&count); err != nil {
		t.Fatalf("count messages: %v", err)
	}
	if count != 0 {
		t.Fatalf("message count = %d, want 0", count)
	}
}