nexarch/dms/internal/migrationdocument/migrate.go

// Package migrationdocument implementiert DOK-01: Uebernahme aller
// Dokumente und Datei-Revisionen aus archivdms in das DMS-Zielschema
// (dms/migrations/tenant/0001_documents.up.sql) — Binaerdaten per
// Objekt-Storage-Kopie ueber dieselbe storage.Service/crypto.Service-
// Kette wie der reguläre Upload-Pfad (dms/internal/upload), Metadaten per
// SQL-Export/Import ueber Source.
package migrationdocument

import (
	"bytes"
	"context"
	"crypto/sha256"
	"errors"
	"fmt"
	"io"

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

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

// SourceDocument ist eine Zeile aus archivdms' Dokumenttabelle (SQL-Export).
type SourceDocument struct {
	SourceDocumentID string
	Title            string
	SourceUserID     string
}

// SourceRevision ist eine Zeile aus archivdms' Revisionstabelle, ObjectKey
// verweist auf die Binaerdatei im Quell-Objekt-Storage.
type SourceRevision struct {
	SourceRevisionID string
	SourceDocumentID string
	RevisionNumber   int
	ChecksumSHA256   string
	SizeBytes        int64
	MimeType         string
	SourceUserID     string
	ObjectKey        string
}

// Source kapselt den lesenden Zugriff auf archivdms: Metadaten per
// SQL-Export (Documents/Revisions), Binaerdaten per Objekt-Storage-Kopie
// (OpenObject). Eine echte Implementierung liest gegen archivdms, Tests
// verwenden ausschliesslich synthetische Fixtures.
type Source interface {
	Documents(ctx context.Context) ([]SourceDocument, error)
	Revisions(ctx context.Context, sourceDocumentID string) ([]SourceRevision, error)
	OpenObject(ctx context.Context, objectKey string) (io.ReadCloser, error)
}

// UserResolver bildet eine archivdms-Benutzer-ID auf die bereits migrierte
// NEXARCH-Benutzer-ID ab (users.id, siehe Voraussetzung in COR-01). ACL-/
// Benutzermigration selbst ist nicht Bestandteil von DOK-01.
type UserResolver interface {
	ResolveUserID(ctx context.Context, sourceUserID string) (string, error)
}

// ErrChecksumMismatch meldet eine Abweichung zwischen der beim Kopieren
// tatsaechlich berechneten Pruefsumme und der von Source gelieferten
// Metadaten-Pruefsumme (Pruefung 1: Hash-Vergleich gegen Quelle).
var ErrChecksumMismatch = errors.New("migrationdocument: pruefsumme weicht von quelle ab")

// Migrator uebertraegt Dokumente+Revisionen einer Quelle in genau einen
// Ziel-Tenant (eine physische Tenant-DB, Modell C).
type Migrator struct {
	pool       *pgxpool.Pool
	source     Source
	users      UserResolver
	storageSvc *storage.Service
	cryptoSvc  *dmscrypto.Service
	tenantSlug string
}

func NewMigrator(pool *pgxpool.Pool, source Source, users UserResolver, storageSvc *storage.Service, cryptoSvc *dmscrypto.Service, tenantSlug string) *Migrator {
	return &Migrator{pool: pool, source: source, users: users, storageSvc: storageSvc, cryptoSvc: cryptoSvc, tenantSlug: tenantSlug}
}

// Result fasst den Ausgang eines Migrationslaufs zusammen.
type Result struct {
	DocumentsMigrated int
	RevisionsMigrated int
	RevisionsSkipped  int
}

// DocumentID ist die Ziel-ID, unter der MigrateAll ein Quelldokument ablegt.
// Nachgelagerte Schritte in anderen Modulen (TAR-01/TAR-02 in archive)
// brauchen sie als ObjectReference, koennen deriveUUID aber nicht importieren.
func DocumentID(tenantSlug, sourceDocumentID string) string {
	return deriveUUID("document", tenantSlug, sourceDocumentID)
}

// MigrateAll uebertraegt alle Dokumente samt Revisionen (Akzeptanzkriterium
// 1). Dokument- und Revisions-IDs im Ziel werden deterministisch aus den
// Quell-IDs abgeleitet (deriveUUID) statt per gen_random_uuid() vergeben —
// ein erneuter Lauf gegen dieselbe Quelle trifft so IMMER dieselben Zeilen
// und erzeugt keine Duplikate (Akzeptanzkriterium 3), ganz ohne eine
// zusaetzliche Mapping-Tabelle.
func (m *Migrator) MigrateAll(ctx context.Context) (Result, error) {
	var res Result
	docs, err := m.source.Documents(ctx)
	if err != nil {
		return res, fmt.Errorf("migrationdocument: dokumente aus quelle lesen: %w", err)
	}

	for _, doc := range docs {
		if err := m.migrateDocument(ctx, doc, &res); err != nil {
			return res, err
		}
	}
	return res, nil
}

func (m *Migrator) migrateDocument(ctx context.Context, doc SourceDocument, res *Result) error {
	documentID := deriveUUID("document", m.tenantSlug, doc.SourceDocumentID)
	createdBy, err := m.users.ResolveUserID(ctx, doc.SourceUserID)
	if err != nil {
		return fmt.Errorf("migrationdocument: ersteller fuer dokument %q aufloesen: %w", doc.SourceDocumentID, err)
	}

	tag, err := m.pool.Exec(ctx, `
		INSERT INTO documents (id, title, created_by)
		VALUES ($1, $2, $3)
		ON CONFLICT (id) DO NOTHING
	`, documentID, doc.Title, createdBy)
	if err != nil {
		return fmt.Errorf("migrationdocument: dokument %q anlegen: %w", doc.SourceDocumentID, err)
	}
	if tag.RowsAffected() > 0 {
		res.DocumentsMigrated++
	}

	revisions, err := m.source.Revisions(ctx, doc.SourceDocumentID)
	if err != nil {
		return fmt.Errorf("migrationdocument: revisionen fuer dokument %q lesen: %w", doc.SourceDocumentID, err)
	}

	var latestRevisionID string
	var latestRevisionNumber int = -1
	for _, rev := range revisions {
		revisionID, err := m.migrateRevision(ctx, documentID, rev, res)
		if err != nil {
			return err
		}
		if rev.RevisionNumber > latestRevisionNumber {
			latestRevisionNumber = rev.RevisionNumber
			latestRevisionID = revisionID
		}
	}

	if latestRevisionID != "" {
		if _, err := m.pool.Exec(ctx, `
			UPDATE documents SET current_revision_id = $2 WHERE id = $1 AND current_revision_id IS DISTINCT FROM $2
		`, documentID, latestRevisionID); err != nil {
			return fmt.Errorf("migrationdocument: aktuelle revision fuer dokument %q setzen: %w", doc.SourceDocumentID, err)
		}
	}
	return nil
}

func (m *Migrator) migrateRevision(ctx context.Context, documentID string, rev SourceRevision, res *Result) (revisionID string, err error) {
	revisionID = deriveUUID("revision", m.tenantSlug, rev.SourceRevisionID)

	var existing bool
	if err := m.pool.QueryRow(ctx, `SELECT true FROM file_revisions WHERE id = $1`, revisionID).Scan(&existing); err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return "", fmt.Errorf("migrationdocument: bestand von revision %q pruefen: %w", rev.SourceRevisionID, err)
	}
	if existing {
		res.RevisionsSkipped++
		return revisionID, nil
	}

	createdBy, err := m.users.ResolveUserID(ctx, rev.SourceUserID)
	if err != nil {
		return "", fmt.Errorf("migrationdocument: ersteller fuer revision %q aufloesen: %w", rev.SourceRevisionID, err)
	}

	plaintext, err := m.source.OpenObject(ctx, rev.ObjectKey)
	if err != nil {
		return "", fmt.Errorf("migrationdocument: binaerobjekt fuer revision %q oeffnen: %w", rev.SourceRevisionID, err)
	}
	defer func() { _ = plaintext.Close() }()

	hasher := sha256.New()
	tee := io.TeeReader(plaintext, hasher)
	env, err := m.cryptoSvc.Seal(ctx, m.tenantSlug, tee)
	if err != nil {
		return "", fmt.Errorf("migrationdocument: revision %q verschluesseln: %w", rev.SourceRevisionID, err)
	}
	ciphertext, err := io.ReadAll(env.Ciphertext)
	if err != nil {
		return "", fmt.Errorf("migrationdocument: chiffretext fuer revision %q lesen: %w", rev.SourceRevisionID, err)
	}

	checksum := fmt.Sprintf("%x", hasher.Sum(nil))
	if checksum != rev.ChecksumSHA256 {
		return "", fmt.Errorf("%w: revision %q (quelle %s, kopiert %s)", ErrChecksumMismatch, rev.SourceRevisionID, rev.ChecksumSHA256, checksum)
	}

	storageKey := storage.ObjectKey(documentID, revisionID)
	if _, err := m.storageSvc.Put(ctx, storageKey, bytes.NewReader(ciphertext), int64(len(ciphertext)), "application/octet-stream"); err != nil {
		return "", fmt.Errorf("migrationdocument: binaerobjekt fuer revision %q ablegen: %w", rev.SourceRevisionID, err)
	}

	if _, err := m.pool.Exec(ctx, `
		INSERT INTO file_revisions (id, document_id, revision_number, storage_key, checksum_sha256, size_bytes, mime_type, created_by, wrapped_dek)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
		ON CONFLICT (id) DO NOTHING
	`, revisionID, documentID, rev.RevisionNumber, storageKey, checksum, rev.SizeBytes, rev.MimeType, createdBy, env.WrappedDEK); err != nil {
		return "", fmt.Errorf("migrationdocument: revision %q anlegen: %w", rev.SourceRevisionID, err)
	}

	res.RevisionsMigrated++
	return revisionID, nil
}