// ARC-08: Tenant-KEK-Rotation. Core (API-10, `internal/kek.Store.
// RotateTenantKEK`) ersetzt den Tenant-KEK durch einen komplett neuen
// Wert — Core selbst hält KEINE Historie vor, `TenantKEKHandler` liefert
// immer nur den AKTUELLEN Schlüssel (siehe kekprovider.go). Damit ARC-08s
// Akzeptanzkriterium 3 ("alte Schlüsselversionen bleiben für
// Lesezugriff kontrolliert verfügbar") erfüllbar ist, muss Mail selbst
// jeden von Core bezogenen Tenant-KEK versioniert zwischenspeichern —
// KEKVersionStore übernimmt genau das, lokal mit einem eigenen,
// ausschließlich über Umgebungsvariable bezogenen Wrap-Schlüssel
// verschlüsselt (kein Klartext-KEK in der Datenbank).
package crypto

import (
	"bytes"
	"context"
	"errors"
	"fmt"

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

// ErrKEKVersionNotFound wird geliefert, wenn die angefragte Version für
// den Mandanten nicht existiert.
var ErrKEKVersionNotFound = errors.New("crypto: kek-version nicht gefunden")

// ErrKEKVersionRevoked wird geliefert, wenn die angefragte Version gezielt
// gesperrt wurde (Pflichtprüfung 2: kompromittierter alter Schlüssel kann
// gezielt gesperrt werden) — der Lesezugriff auf mit dieser Version
// verschlüsselte Altobjekte ist dann bewusst blockiert.
var ErrKEKVersionRevoked = errors.New("crypto: kek-version wurde gesperrt")

// KEKVersionStore verwaltet die Versionshistorie der Tenant-KEKs, die
// dieses Mail-Modul im Lauf der Zeit von Core bezogen hat.
type KEKVersionStore struct {
	pool         *pgxpool.Pool
	localWrapKey []byte
}

// NewKEKVersionStore erzeugt einen Store. localWrapKey verschlüsselt die
// zwischengespeicherten Tenant-KEKs lokal at rest (KEKSize Bytes,
// ausschließlich über Umgebungsvariable bezogen — nie im Code).
func NewKEKVersionStore(pool *pgxpool.Pool, localWrapKey []byte) *KEKVersionStore {
	return &KEKVersionStore{pool: pool, localWrapKey: localWrapKey}
}

// EnsureSchema legt die Tabelle an, falls sie noch nicht existiert —
// gleiches Muster wie mail/internal/dedup/indexworker (kein zentraler
// Migrationsläufer für Mandanten-Datenbanken im Mail-Modul vorhanden).
func (s *KEKVersionStore) EnsureSchema(ctx context.Context) error {
	if _, err := s.pool.Exec(ctx, `
		CREATE TABLE IF NOT EXISTS mail_kek_versions (
			tenant_slug  TEXT NOT NULL,
			version      INT NOT NULL,
			wrapped_kek  BYTEA NOT NULL,
			revoked      BOOLEAN NOT NULL DEFAULT false,
			created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
			PRIMARY KEY (tenant_slug, version)
		)
	`); err != nil {
		return fmt.Errorf("crypto: kek-versionsschema anlegen: %w", err)
	}
	return nil
}

// RecordIfNew merkt sich plainKEK als neue Version für tenantSlug, FALLS
// er sich vom zuletzt gespeicherten Wert unterscheidet (Rotation
// erkannt) — bei unverändertem KEK wird keine neue Version angelegt,
// sondern die bestehende Versionsnummer zurückgegeben (Akzeptanzkriterium
// 1: rotierbar verwaltet, nicht bei jedem Aufruf eine neue Version).
func (s *KEKVersionStore) RecordIfNew(ctx context.Context, tenantSlug string, plainKEK []byte) (version int, err error) {
	var latestVersion int
	var latestWrapped []byte
	err = s.pool.QueryRow(ctx, `
		SELECT version, wrapped_kek FROM mail_kek_versions
		WHERE tenant_slug = $1 ORDER BY version DESC LIMIT 1
	`, tenantSlug).Scan(&latestVersion, &latestWrapped)
	switch {
	case errors.Is(err, pgx.ErrNoRows):
		return s.insertVersion(ctx, tenantSlug, 1, plainKEK)
	case err != nil:
		return 0, fmt.Errorf("crypto: letzte kek-version lesen: %w", err)
	}

	latestPlain, err := open(s.localWrapKey, latestWrapped)
	if err != nil {
		return 0, fmt.Errorf("crypto: zwischengespeicherten kek entpacken: %w", err)
	}
	if bytes.Equal(latestPlain, plainKEK) {
		return latestVersion, nil
	}
	return s.insertVersion(ctx, tenantSlug, latestVersion+1, plainKEK)
}

func (s *KEKVersionStore) insertVersion(ctx context.Context, tenantSlug string, version int, plainKEK []byte) (int, error) {
	wrapped, err := seal(s.localWrapKey, plainKEK)
	if err != nil {
		return 0, fmt.Errorf("crypto: kek für zwischenspeicherung verpacken: %w", err)
	}
	if _, err := s.pool.Exec(ctx, `
		INSERT INTO mail_kek_versions (tenant_slug, version, wrapped_kek) VALUES ($1, $2, $3)
	`, tenantSlug, version, wrapped); err != nil {
		return 0, fmt.Errorf("crypto: kek-version speichern: %w", err)
	}
	return version, nil
}

// Get liefert den entschlüsselten historischen Tenant-KEK einer
// bestimmten Version. Liefert ErrKEKVersionRevoked, wenn die Version
// gezielt gesperrt wurde (Pflichtprüfung 2).
func (s *KEKVersionStore) Get(ctx context.Context, tenantSlug string, version int) ([]byte, error) {
	var wrapped []byte
	var revoked bool
	err := s.pool.QueryRow(ctx, `
		SELECT wrapped_kek, revoked FROM mail_kek_versions
		WHERE tenant_slug = $1 AND version = $2
	`, tenantSlug, version).Scan(&wrapped, &revoked)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, ErrKEKVersionNotFound
		}
		return nil, fmt.Errorf("crypto: kek-version lesen: %w", err)
	}
	if revoked {
		return nil, ErrKEKVersionRevoked
	}
	return open(s.localWrapKey, wrapped)
}

// Revoke sperrt eine Tenant-KEK-Version gezielt (Pflichtprüfung 2):
// nachfolgende Get-Aufrufe für genau diese Version schlagen mit
// ErrKEKVersionRevoked fehl, andere Versionen bleiben unberührt.
func (s *KEKVersionStore) Revoke(ctx context.Context, tenantSlug string, version int) error {
	tag, err := s.pool.Exec(ctx, `
		UPDATE mail_kek_versions SET revoked = true WHERE tenant_slug = $1 AND version = $2
	`, tenantSlug, version)
	if err != nil {
		return fmt.Errorf("crypto: kek-version sperren: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return ErrKEKVersionNotFound
	}
	return nil
}