nexarch/workflow/internal/migrationworkflow/migrate.go

// Package migrationworkflow implementiert WFL-01: Uebernahme abgeschlossener
// Workflow-/Automatisierungsvorgaenge aus archivdms als Historie nach
// ENG-01 (workflow_instances/workflow_instance_history). Laufende Instanzen
// werden NICHT migriert, sie muessen vor dem Stichtag im Altsystem zu Ende
// gefuehrt oder abgebrochen werden.
package migrationworkflow

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

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

// Status ist der Abschlusszustand eines Vorgangs in der Quelle.
type Status string

const (
	StatusCompleted Status = "completed"
	StatusAborted   Status = "aborted"
	StatusRunning   Status = "running"
)

// done listet die Status-Werte, die migriert werden duerfen (Whitelist statt
// Blacklist: ein unbekannter/neuer Status-Wert wird NICHT migriert, siehe
// migrateDone).
var done = map[Status]bool{
	StatusCompleted: true,
	StatusAborted:   true,
}

// SourceCase ist ein Vorgang (Workflow-/Automatisierungsinstanz) aus
// archivdms.
type SourceCase struct {
	SourceCaseID string
	ProcessName  string
	FinalState   string
	Status       Status
}

// SourceHistoryEntry ist ein Historieneintrag eines Vorgangs aus archivdms,
// in Reihenfolge ihres Auftretens.
type SourceHistoryEntry struct {
	FromState         string
	ToState           string
	SourceActor       string
	Reason            string
	OccurredAtRFC3339 string
}

// Source kapselt den lesenden Zugriff auf archivdms (SQL-Export). Eine
// echte Implementierung liest gegen archivdms, Tests verwenden
// ausschliesslich synthetische Fixtures.
type Source interface {
	Cases(ctx context.Context) ([]SourceCase, error)
	History(ctx context.Context, sourceCaseID string) ([]SourceHistoryEntry, error)
}

// UserResolver bildet eine archivdms-Akteurs-ID auf die bereits migrierte
// NEXARCH-Benutzer-ID ab (siehe COR-02, user_migration_map). Benutzer-/
// ACL-Migration selbst ist nicht Bestandteil von WFL-01.
type UserResolver interface {
	ResolveUserID(ctx context.Context, sourceActorID string) (string, error)
}

// ErrUnresolvedActor wird zurueckgegeben, wenn ein Akteur in einem
// Historieneintrag nicht auf eine NEXARCH-Benutzer-ID abgebildet werden kann
// (fail-closed statt stillem Platzhalter).
var ErrUnresolvedActor = errors.New("migrationworkflow: akteur nicht aufloesbar")

// Migrator uebertraegt abgeschlossene Vorgaenge einer Quelle als Historie
// in Workflow ENG-01.
type Migrator struct {
	pool   *pgxpool.Pool
	source Source
	users  UserResolver
}

func NewMigrator(pool *pgxpool.Pool, source Source, users UserResolver) *Migrator {
	return &Migrator{pool: pool, source: source, users: users}
}

// Result fasst den Ausgang eines Migrationslaufs zusammen.
type Result struct {
	InstancesMigrated        int
	InstancesSkippedExisting int
	InstancesSkippedOpen     int
	HistoryEntriesMigrated   int
}

// instanceID leitet eine deterministische Ziel-ID aus der Quell-ID ab, damit
// ein erneuter Lauf gegen dieselbe Quelle dieselbe Zeile trifft (keine
// Duplikate, Akzeptanzkriterium 2/Pruefung 1).
func instanceID(sourceCaseID string) string {
	h := sha256.Sum256([]byte("workflow_instance\x00" + sourceCaseID))
	h[6] = (h[6] & 0x0f) | 0x50
	h[8] = (h[8] & 0x3f) | 0x80
	return fmt.Sprintf("%x-%x-%x-%x-%x", h[0:4], h[4:6], h[6:8], h[8:10], h[10:16])
}

// MigrateCompleted uebertraegt jeden abgeschlossenen Vorgang (Status
// completed/aborted) samt vollstaendiger Historie (Akzeptanzkriterium 1). Ein
// laufender Vorgang (oder ein unbekannter, nicht in der Whitelist gefuehrter
// Status) wird NICHT angefasst (Akzeptanzkriterium 2, sicherheitskritischer
// Teil dieser Kachel).
func (m *Migrator) MigrateCompleted(ctx context.Context) (Result, error) {
	var res Result
	cases, err := m.source.Cases(ctx)
	if err != nil {
		return res, fmt.Errorf("migrationworkflow: vorgaenge aus quelle lesen: %w", err)
	}

	for _, c := range cases {
		if !done[c.Status] {
			res.InstancesSkippedOpen++
			continue
		}
		if err := m.migrateCase(ctx, c, &res); err != nil {
			return res, err
		}
	}
	return res, nil
}

func (m *Migrator) migrateCase(ctx context.Context, c SourceCase, res *Result) error {
	id := instanceID(c.SourceCaseID)

	var alreadyMigrated bool
	if err := m.pool.QueryRow(ctx, `SELECT true FROM workflow_instances WHERE id = $1`, id).Scan(&alreadyMigrated); err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return fmt.Errorf("migrationworkflow: bestand von instanz %q pruefen: %w", c.SourceCaseID, err)
	}
	if alreadyMigrated {
		res.InstancesSkippedExisting++
		return nil
	}

	history, err := m.source.History(ctx, c.SourceCaseID)
	if err != nil {
		return fmt.Errorf("migrationworkflow: historie fuer vorgang %q lesen: %w", c.SourceCaseID, err)
	}

	resolved := make([]struct {
		entry SourceHistoryEntry
		actor string
	}, len(history))
	for i, h := range history {
		actorID, err := m.users.ResolveUserID(ctx, h.SourceActor)
		if err != nil {
			return fmt.Errorf("%w: vorgang %q, akteur %q: %v", ErrUnresolvedActor, c.SourceCaseID, h.SourceActor, err)
		}
		resolved[i].entry = h
		resolved[i].actor = actorID
	}

	version := len(history) + 1
	if _, err := m.pool.Exec(ctx, `
		INSERT INTO workflow_instances (id, process_name, state, version)
		VALUES ($1, $2, $3, $4)
		ON CONFLICT (id) DO NOTHING
	`, id, c.ProcessName, c.FinalState, version); err != nil {
		return fmt.Errorf("migrationworkflow: instanz %q anlegen: %w", c.SourceCaseID, err)
	}
	res.InstancesMigrated++

	for _, r := range resolved {
		if _, err := m.pool.Exec(ctx, `
			INSERT INTO workflow_instance_history (instance_id, from_state, to_state, actor, reason, occurred_at)
			VALUES ($1, $2, $3, $4, $5, $6)
		`, id, r.entry.FromState, r.entry.ToState, r.actor, r.entry.Reason, r.entry.OccurredAtRFC3339); err != nil {
			return fmt.Errorf("migrationworkflow: historieneintrag fuer instanz %q schreiben: %w", c.SourceCaseID, err)
		}
		res.HistoryEntriesMigrated++
	}
	return nil
}

// OpenCases liefert alle Vorgaenge der Quelle, die NICHT abgeschlossen sind
// (Status ausserhalb der Whitelist, insbesondere laufend). Grundlage fuer
// die Pruefung vor dem Cutover (Akzeptanzkriterium 3/Pruefung 2): diese
// Liste muss vor dem Stichtag leer sein oder bewusst abgearbeitet werden.
func (m *Migrator) OpenCases(ctx context.Context) ([]SourceCase, error) {
	cases, err := m.source.Cases(ctx)
	if err != nil {
		return nil, fmt.Errorf("migrationworkflow: vorgaenge aus quelle lesen: %w", err)
	}
	var open []SourceCase
	for _, c := range cases {
		if !done[c.Status] {
			open = append(open, c)
		}
	}
	return open, nil
}