nexarch/dms/internal/migrationdocument/migrate_test.go

package migrationdocument

import (
	"bytes"
	"context"
	"crypto/sha256"
	"errors"
	"fmt"
	"io"
	"os"
	"testing"
	"time"

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

	"gitea.perlbach24.de/scripte/nexarch/common/testdb"
	dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/storage"
)

// fakeKEKProvider liefert einen fest hinterlegten Tenant-KEK, analog zum
// Fake in dms/internal/upload (dort nicht exportiert).
type fakeKEKProvider struct{ kek []byte }

func (f *fakeKEKProvider) TenantKEK(ctx context.Context, tenantSlug string) ([]byte, error) {
	return f.kek, nil
}

type noopUsageReporter struct{}

func (noopUsageReporter) Report(ctx context.Context, tenantSlug, metric string, delta int64) error {
	return nil
}

// fixtureObject ist ein SYNTHETISCHES Dokument fuer die Testquelle — fake
// Inhalt, keine echten archivdms-Daten (Produktarchitektur-Vorgabe).
type fixtureObject struct {
	content []byte
}

func sha256Hex(b []byte) string {
	sum := sha256.Sum256(b)
	return fmt.Sprintf("%x", sum[:])
}

// memorySource ist eine In-Memory-Testquelle: simuliert das SQL-Export/
// Objekt-Storage-Paar von archivdms rein synthetisch.
type memorySource struct {
	docs      []SourceDocument
	revisions map[string][]SourceRevision
	objects   map[string]fixtureObject
}

func (s *memorySource) Documents(ctx context.Context) ([]SourceDocument, error) {
	return s.docs, nil
}

func (s *memorySource) Revisions(ctx context.Context, sourceDocumentID string) ([]SourceRevision, error) {
	return s.revisions[sourceDocumentID], nil
}

func (s *memorySource) OpenObject(ctx context.Context, objectKey string) (io.ReadCloser, error) {
	obj, ok := s.objects[objectKey]
	if !ok {
		return nil, fmt.Errorf("fixture-objekt %q nicht gefunden", objectKey)
	}
	return io.NopCloser(bytes.NewReader(obj.content)), nil
}

// staticUserResolver bildet archivdms-Benutzer-IDs 1:1 auf eine feste
// Ziel-Benutzer-ID ab (Benutzer-/ACL-Migration ist nicht Bestandteil
// von DOK-01).
type staticUserResolver struct{ userID string }

func (r staticUserResolver) ResolveUserID(ctx context.Context, sourceUserID string) (string, error) {
	return r.userID, nil
}

func newSyntheticSource() *memorySource {
	return &memorySource{
		docs: []SourceDocument{
			{SourceDocumentID: "archivdms-doc-1", Title: "Testakte Musterfall.pdf", SourceUserID: "archivdms-user-1"},
			{SourceDocumentID: "archivdms-doc-2", Title: "Testakte Zweitfall.txt", SourceUserID: "archivdms-user-1"},
		},
		revisions: map[string][]SourceRevision{
			"archivdms-doc-1": {
				{SourceRevisionID: "archivdms-doc-1-rev-1", SourceDocumentID: "archivdms-doc-1", RevisionNumber: 1, SizeBytes: int64(len("Fake-PDF-Inhalt-V1")), MimeType: "application/pdf", SourceUserID: "archivdms-user-1", ObjectKey: "doc-1-rev-1", ChecksumSHA256: sha256Hex([]byte("Fake-PDF-Inhalt-V1"))},
				{SourceRevisionID: "archivdms-doc-1-rev-2", SourceDocumentID: "archivdms-doc-1", RevisionNumber: 2, SizeBytes: int64(len("Fake-PDF-Inhalt-V2-ueberarbeitet")), MimeType: "application/pdf", SourceUserID: "archivdms-user-1", ObjectKey: "doc-1-rev-2", ChecksumSHA256: sha256Hex([]byte("Fake-PDF-Inhalt-V2-ueberarbeitet"))},
			},
			"archivdms-doc-2": {
				{SourceRevisionID: "archivdms-doc-2-rev-1", SourceDocumentID: "archivdms-doc-2", RevisionNumber: 1, SizeBytes: int64(len("Fake-Text-Inhalt")), MimeType: "text/plain", SourceUserID: "archivdms-user-1", ObjectKey: "doc-2-rev-1", ChecksumSHA256: sha256Hex([]byte("Fake-Text-Inhalt"))},
			},
		},
		objects: map[string]fixtureObject{
			"doc-1-rev-1": {content: []byte("Fake-PDF-Inhalt-V1")},
			"doc-1-rev-2": {content: []byte("Fake-PDF-Inhalt-V2-ueberarbeitet")},
			"doc-2-rev-1": {content: []byte("Fake-Text-Inhalt")},
		},
	}
}

func setupMigratorTest(t *testing.T) (*Migrator, *pgxpool.Pool, *storage.LocalDriver, string) {
	t.Helper()
	dsn := os.Getenv("TEST_TENANT_DSN")
	if dsn == "" {
		t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen")
	}
	ctx := context.Background()

	pool := testdb.Pool(t, dsn)
	t.Cleanup(func() { pool.Close() })

	if _, err := pool.Exec(ctx, `
		CREATE EXTENSION IF NOT EXISTS pgcrypto;
		CREATE TABLE IF NOT EXISTS users (
			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), email TEXT NOT NULL UNIQUE, name TEXT NOT NULL,
			created_at TIMESTAMPTZ NOT NULL DEFAULT now()
		);
		CREATE TABLE IF NOT EXISTS documents (
			id UUID PRIMARY KEY, folder_id UUID, title TEXT NOT NULL, current_revision_id UUID,
			created_by UUID NOT NULL REFERENCES users(id),
			created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), deleted_at TIMESTAMPTZ
		);
		CREATE TABLE IF NOT EXISTS file_revisions (
			id UUID PRIMARY KEY, document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
			revision_number INT NOT NULL, storage_key TEXT NOT NULL, checksum_sha256 TEXT NOT NULL,
			size_bytes BIGINT NOT NULL, mime_type TEXT NOT NULL, created_by UUID NOT NULL REFERENCES users(id),
			created_at TIMESTAMPTZ NOT NULL DEFAULT now(), wrapped_dek BYTEA,
			confirmed_duplicate BOOLEAN NOT NULL DEFAULT false,
			UNIQUE (document_id, revision_number)
		);
		ALTER TABLE documents DROP CONSTRAINT IF EXISTS fk_documents_current_revision;
		ALTER TABLE documents ADD CONSTRAINT fk_documents_current_revision
			FOREIGN KEY (current_revision_id) REFERENCES file_revisions(id) ON DELETE SET NULL;
	`); err != nil {
		t.Fatalf("schema: %v", err)
	}
	t.Cleanup(func() {
		ctx := context.Background()
		_, _ = pool.Exec(ctx, `TRUNCATE file_revisions, documents, users CASCADE`)
	})

	var userID string
	if err := pool.QueryRow(ctx, `
		INSERT INTO users (email, name) VALUES ($1, 'Migrations-Testbenutzer') RETURNING id
	`, fmt.Sprintf("dok01-test-%d@example.test", time.Now().UnixNano())).Scan(&userID); err != nil {
		t.Fatalf("testbenutzer anlegen: %v", err)
	}

	driver := storage.NewLocalDriver(t.TempDir(), []byte("secret"), "https://files.example.test", 0)
	storageSvc := storage.NewService(driver, noopUsageReporter{}, "acme-test")
	cryptoSvc := dmscrypto.NewService(&fakeKEKProvider{kek: bytes.Repeat([]byte{0x22}, dmscrypto.KEKSize)})

	migrator := NewMigrator(pool, newSyntheticSource(), staticUserResolver{userID: userID}, storageSvc, cryptoSvc, "acme-test")
	return migrator, pool, driver, userID
}

// TestMigrateAll_AkzeptanzkriteriumVollstaendigReferenzierbar prueft
// Akzeptanzkriterium 1: jedes Dokument samt aller Revisionen ist nach der
// Migration vollstaendig referenzierbar.
func TestMigrateAll_AkzeptanzkriteriumVollstaendigReferenzierbar(t *testing.T) {
	migrator, pool, _, _ := setupMigratorTest(t)
	ctx := context.Background()

	res, err := migrator.MigrateAll(ctx)
	if err != nil {
		t.Fatalf("MigrateAll: %v", err)
	}
	if res.DocumentsMigrated != 2 || res.RevisionsMigrated != 3 {
		t.Fatalf("erwartet 2 dokumente/3 revisionen, habe %+v", res)
	}

	var documentCount, revisionCount int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM documents`).Scan(&documentCount); err != nil {
		t.Fatalf("dokumente zaehlen: %v", err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM file_revisions`).Scan(&revisionCount); err != nil {
		t.Fatalf("revisionen zaehlen: %v", err)
	}
	if documentCount != 2 || revisionCount != 3 {
		t.Fatalf("erwartet 2 dokumente/3 revisionen in der db, habe %d/%d", documentCount, revisionCount)
	}

	var revisionsForDoc1 int
	documentID1 := deriveUUID("document", "acme-test", "archivdms-doc-1")
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM file_revisions WHERE document_id = $1`, documentID1).Scan(&revisionsForDoc1); err != nil {
		t.Fatalf("revisionen fuer dokument 1 zaehlen: %v", err)
	}
	if revisionsForDoc1 != 2 {
		t.Fatalf("erwartet 2 revisionen fuer dokument 1, habe %d", revisionsForDoc1)
	}
}

// TestMigrateAll_AkzeptanzkriteriumKonsistenteVerknuepfung prueft
// Akzeptanzkriterium 2: Binaerdaten und Metadaten sind konsistent
// miteinander verknuepft — current_revision_id zeigt auf die hoechste
// Revisionsnummer, und das Objekt unter storage_key ist tatsaechlich
// abrufbar mit identischem Inhalt (Pruefung 1: Hash-Vergleich).
func TestMigrateAll_AkzeptanzkriteriumKonsistenteVerknuepfung(t *testing.T) {
	migrator, pool, driver, _ := setupMigratorTest(t)
	ctx := context.Background()

	if _, err := migrator.MigrateAll(ctx); err != nil {
		t.Fatalf("MigrateAll: %v", err)
	}

	documentID1 := deriveUUID("document", "acme-test", "archivdms-doc-1")
	var currentRevisionID, storageKey string
	if err := pool.QueryRow(ctx, `
		SELECT d.current_revision_id, r.storage_key
		FROM documents d JOIN file_revisions r ON r.id = d.current_revision_id
		WHERE d.id = $1
	`, documentID1).Scan(&currentRevisionID, &storageKey); err != nil {
		t.Fatalf("aktuelle revision lesen: %v", err)
	}
	expectedRevisionID := deriveUUID("revision", "acme-test", "archivdms-doc-1-rev-2")
	if currentRevisionID != expectedRevisionID {
		t.Fatalf("erwartet current_revision_id %q (hoechste revisionsnummer), habe %q", expectedRevisionID, currentRevisionID)
	}

	rc, err := driver.Get(ctx, storageKey)
	if err != nil {
		t.Fatalf("objekt unter storage_key lesen: %v", err)
	}
	defer rc.Close()
	ciphertext, err := io.ReadAll(rc)
	if err != nil {
		t.Fatalf("ciphertext lesen: %v", err)
	}

	var wrappedDEK []byte
	if err := pool.QueryRow(ctx, `SELECT wrapped_dek FROM file_revisions WHERE id = $1`, currentRevisionID).Scan(&wrappedDEK); err != nil {
		t.Fatalf("wrapped_dek lesen: %v", err)
	}
	cryptoSvc := dmscrypto.NewService(&fakeKEKProvider{kek: bytes.Repeat([]byte{0x22}, dmscrypto.KEKSize)})
	plainReader, err := cryptoSvc.Open(ctx, "acme-test", wrappedDEK, bytes.NewReader(ciphertext))
	if err != nil {
		t.Fatalf("entschluesseln: %v", err)
	}
	plaintext, err := io.ReadAll(plainReader)
	if err != nil {
		t.Fatalf("klartext lesen: %v", err)
	}
	if string(plaintext) != "Fake-PDF-Inhalt-V2-ueberarbeitet" {
		t.Fatalf("entschluesselter inhalt weicht von quelle ab: %q", plaintext)
	}
	if sha256Hex(plaintext) != sha256Hex([]byte("Fake-PDF-Inhalt-V2-ueberarbeitet")) {
		t.Fatalf("hash-vergleich gegen quelle fehlgeschlagen")
	}
}

// TestMigrateAll_AkzeptanzkriteriumWiederholbar prueft Akzeptanzkriterium
// 3: ein erneuter Lauf gegen dieselbe Quelle erzeugt keine Duplikate.
func TestMigrateAll_AkzeptanzkriteriumWiederholbar(t *testing.T) {
	migrator, pool, _, _ := setupMigratorTest(t)
	ctx := context.Background()

	if _, err := migrator.MigrateAll(ctx); err != nil {
		t.Fatalf("erster lauf: %v", err)
	}
	res2, err := migrator.MigrateAll(ctx)
	if err != nil {
		t.Fatalf("zweiter lauf: %v", err)
	}
	if res2.DocumentsMigrated != 0 || res2.RevisionsMigrated != 0 || res2.RevisionsSkipped != 3 {
		t.Fatalf("zweiter lauf haette nichts neu migrieren duerfen, habe %+v", res2)
	}

	var documentCount, revisionCount int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM documents`).Scan(&documentCount); err != nil {
		t.Fatalf("dokumente zaehlen: %v", err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM file_revisions`).Scan(&revisionCount); err != nil {
		t.Fatalf("revisionen zaehlen: %v", err)
	}
	if documentCount != 2 || revisionCount != 3 {
		t.Fatalf("erneuter lauf hat duplikate erzeugt: %d dokumente/%d revisionen", documentCount, revisionCount)
	}
}

// TestMigrateRevision_ChecksumMismatch prueft, dass eine zwischen Quelle und
// kopiertem Objekt abweichende Pruefsumme die Migration dieser Revision
// mit einem klaren Fehler abbricht statt sie still zu uebernehmen.
func TestMigrateRevision_ChecksumMismatch(t *testing.T) {
	migrator, _, _, _ := setupMigratorTest(t)
	ctx := context.Background()

	src := migrator.source.(*memorySource)
	src.objects["doc-2-rev-1"] = fixtureObject{content: []byte("manipulierter-inhalt")}

	_, err := migrator.MigrateAll(ctx)
	if !errors.Is(err, ErrChecksumMismatch) {
		t.Fatalf("erwartet ErrChecksumMismatch, habe %v", err)
	}
}