nexarch/dms/internal/textcrypt/migrate.go

package textcrypt

import (
	"context"
	"fmt"

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

// Stats zaehlt Zeilen je Textspeicher.
type Stats struct {
	OCRPages  int
	TextIndex int
	EInvoice  int
}

func (s Stats) Total() int { return s.OCRPages + s.TextIndex + s.EInvoice }

// CountLegacy zaehlt Zeilen, deren Text noch im Klartext liegt.
func CountLegacy(ctx context.Context, pool *pgxpool.Pool) (Stats, error) {
	var s Stats
	err := pool.QueryRow(ctx, `
		SELECT (SELECT count(*) FROM revision_ocr_pages WHERE text_enc IS NULL AND text <> ''),
		       (SELECT count(*) FROM revision_text_index WHERE text_enc IS NULL AND text <> ''),
		       (SELECT count(*) FROM revision_einvoice WHERE search_text_enc IS NULL AND search_text <> '')
	`).Scan(&s.OCRPages, &s.TextIndex, &s.EInvoice)
	return s, err
}

// CountEncrypted zaehlt verschluesselte Zeilen.
func CountEncrypted(ctx context.Context, pool *pgxpool.Pool) (Stats, error) {
	var s Stats
	err := pool.QueryRow(ctx, `
		SELECT (SELECT count(*) FROM revision_ocr_pages WHERE text_enc IS NOT NULL),
		       (SELECT count(*) FROM revision_text_index WHERE text_enc IS NOT NULL),
		       (SELECT count(*) FROM revision_einvoice WHERE search_text_enc IS NOT NULL)
	`).Scan(&s.OCRPages, &s.TextIndex, &s.EInvoice)
	return s, err
}

// MigrateLegacy verschluesselt vorhandene Klartextzeilen. Je Revision eine Transaktion, idempotent:
// ein zweiter Lauf findet nichts mehr.
func MigrateLegacy(ctx context.Context, pool *pgxpool.Pool, v Vault) (Stats, error) {
	var done Stats
	if err := v.ready(); err != nil {
		return done, err
	}
	pages, err := pool.Query(ctx, `SELECT DISTINCT revision_id::text FROM revision_ocr_pages WHERE text_enc IS NULL AND text <> ''`)
	if err != nil {
		return done, err
	}
	revs, err := pgx.CollectRows(pages, pgx.RowTo[string])
	if err != nil {
		return done, err
	}
	for _, rev := range revs {
		n, err := migrateOCRRevision(ctx, pool, v, rev)
		done.OCRPages += n
		if err != nil {
			return done, fmt.Errorf("textcrypt: ocr-revision %s: %w", rev, err)
		}
	}
	n, err := migrateSingle(ctx, pool, v, "revision_text_index", "text", "text_enc", "wrapped_dek", AADIndex)
	done.TextIndex = n
	if err != nil {
		return done, err
	}
	n, err = migrateSingle(ctx, pool, v, "revision_einvoice", "search_text", "search_text_enc", "wrapped_dek", AADInvoice)
	done.EInvoice = n
	return done, err
}

func migrateOCRRevision(ctx context.Context, pool *pgxpool.Pool, v Vault, rev string) (int, error) {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return 0, err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	dek, err := OCRKey(ctx, tx, v, rev)
	if err != nil {
		return 0, err
	}
	rows, err := tx.Query(ctx, `SELECT page, text FROM revision_ocr_pages WHERE revision_id = $1 AND text_enc IS NULL AND text <> '' FOR UPDATE`, rev)
	if err != nil {
		return 0, err
	}
	type row struct {
		page int
		text string
	}
	todo, err := pgx.CollectRows(rows, func(r pgx.CollectableRow) (row, error) {
		var x row
		return x, r.Scan(&x.page, &x.text)
	})
	if err != nil {
		return 0, err
	}
	for _, r := range todo {
		sealed, err := Seal(dek, AADPage(rev, r.page), r.text)
		if err != nil {
			return 0, err
		}
		if _, err := tx.Exec(ctx, `UPDATE revision_ocr_pages SET text_enc = $3, text = '' WHERE revision_id = $1 AND page = $2`, rev, r.page, sealed); err != nil {
			return 0, err
		}
	}
	return len(todo), tx.Commit(ctx)
}

func migrateSingle(ctx context.Context, pool *pgxpool.Pool, v Vault, table, textCol, encCol, keyCol string, aad func(string) []byte) (int, error) {
	rows, err := pool.Query(ctx, fmt.Sprintf(`SELECT revision_id::text, %s FROM %s WHERE %s IS NULL AND %s <> ''`, textCol, table, encCol, textCol))
	if err != nil {
		return 0, err
	}
	type row struct{ rev, text string }
	todo, err := pgx.CollectRows(rows, func(r pgx.CollectableRow) (row, error) {
		var x row
		return x, r.Scan(&x.rev, &x.text)
	})
	if err != nil {
		return 0, err
	}
	n := 0
	for _, r := range todo {
		dek, wrapped, err := v.NewKey(ctx)
		if err != nil {
			return n, err
		}
		sealed, err := Seal(dek, aad(r.rev), r.text)
		if err != nil {
			return n, err
		}
		tag, err := pool.Exec(ctx, fmt.Sprintf(`UPDATE %s SET %s = $2, %s = $3, %s = '' WHERE revision_id = $1 AND %s IS NULL`, table, encCol, keyCol, textCol, encCol), r.rev, sealed, wrapped)
		if err != nil {
			return n, fmt.Errorf("textcrypt: %s %s: %w", table, r.rev, err)
		}
		n += int(tag.RowsAffected())
	}
	return n, nil
}

// DecryptAll stellt den Klartext wieder her (Voraussetzung fuer das Zurueckrollen der Migration 0031).
func DecryptAll(ctx context.Context, pool *pgxpool.Pool, v Vault) (Stats, error) {
	var done Stats
	if err := v.ready(); err != nil {
		return done, err
	}
	revs, err := collectStrings(ctx, pool, `SELECT DISTINCT revision_id::text FROM revision_ocr_pages WHERE text_enc IS NOT NULL`)
	if err != nil {
		return done, err
	}
	for _, rev := range revs {
		var wrapped []byte
		if err := pool.QueryRow(ctx, `SELECT text_wrapped_dek FROM revision_ocr WHERE revision_id = $1`, rev).Scan(&wrapped); err != nil {
			return done, err
		}
		dek, err := v.Key(ctx, wrapped)
		if err != nil {
			return done, err
		}
		rows, err := pool.Query(ctx, `SELECT page, text_enc FROM revision_ocr_pages WHERE revision_id = $1 AND text_enc IS NOT NULL`, rev)
		if err != nil {
			return done, err
		}
		type row struct {
			page int
			enc  []byte
		}
		todo, err := pgx.CollectRows(rows, func(r pgx.CollectableRow) (row, error) {
			var x row
			return x, r.Scan(&x.page, &x.enc)
		})
		if err != nil {
			return done, err
		}
		for _, r := range todo {
			text, err := Open(dek, AADPage(rev, r.page), r.enc)
			if err != nil {
				return done, fmt.Errorf("textcrypt: ocr-revision %s seite %d: %w", rev, r.page, err)
			}
			if _, err := pool.Exec(ctx, `UPDATE revision_ocr_pages SET text = $3, text_enc = NULL WHERE revision_id = $1 AND page = $2`, rev, r.page, text); err != nil {
				return done, err
			}
			done.OCRPages++
		}
	}
	if done.TextIndex, err = decryptSingle(ctx, pool, v, "revision_text_index", "text", "text_enc", "wrapped_dek", AADIndex); err != nil {
		return done, err
	}
	done.EInvoice, err = decryptSingle(ctx, pool, v, "revision_einvoice", "search_text", "search_text_enc", "wrapped_dek", AADInvoice)
	return done, err
}

func collectStrings(ctx context.Context, pool *pgxpool.Pool, sql string) ([]string, error) {
	rows, err := pool.Query(ctx, sql)
	if err != nil {
		return nil, err
	}
	return pgx.CollectRows(rows, pgx.RowTo[string])
}

func decryptSingle(ctx context.Context, pool *pgxpool.Pool, v Vault, table, textCol, encCol, keyCol string, aad func(string) []byte) (int, error) {
	rows, err := pool.Query(ctx, fmt.Sprintf(`SELECT revision_id::text, %s, %s FROM %s WHERE %s IS NOT NULL`, encCol, keyCol, table, encCol))
	if err != nil {
		return 0, err
	}
	type row struct {
		rev     string
		enc, kw []byte
	}
	todo, err := pgx.CollectRows(rows, func(r pgx.CollectableRow) (row, error) {
		var x row
		return x, r.Scan(&x.rev, &x.enc, &x.kw)
	})
	if err != nil {
		return 0, err
	}
	n := 0
	for _, r := range todo {
		dek, err := v.Key(ctx, r.kw)
		if err != nil {
			return n, err
		}
		text, err := Open(dek, aad(r.rev), r.enc)
		if err != nil {
			return n, fmt.Errorf("textcrypt: %s %s: %w", table, r.rev, err)
		}
		if _, err := pool.Exec(ctx, fmt.Sprintf(`UPDATE %s SET %s = $2, %s = NULL, %s = NULL WHERE revision_id = $1`, table, textCol, encCol, keyCol), r.rev, text); err != nil {
			return n, err
		}
		n++
	}
	return n, nil
}