ARC-01: objekt-speicher-anbindung-fuer-mails-anhaenge

- mail/internal/storage: LocalDriver/S3Driver (bewaehrtes Muster aus
  DMS FDN-03, bewusste Neuimplementierung - Mail kann DMS nicht
  importieren), ObjectKey mit festem Pfadschema
- Service.Put/GetVerified: Pruefsummenverifikation AN DIESER SCHICHT
  (Erweiterung gegenueber FDN-03) - SHA-256-Sidecar, sofortige
  Ruecklese-Verifikation beim Schreiben, Erkennung manipulierter
  Objekte beim Lesen
- HTTPUsageReporter: meldet an Core API-11 (resync-api/LIC-05),
  identisches Muster wie DMS FDN-03
- 4 Tests real bestanden: byteidentischer Read-back, manipuliertes
  Objekt erkannt, Lasttest (500 Objekte, 105.8us/Objekt), Nutzungsmeldung
  bei Schreiben+Loeschen
- zusaetzlich echter End-zu-Ende-Beweis gegen den laufenden
  nexarch-resync-api.service: reales Service-Credential provisioniert,
  Put->GetVerified->Delete komplett durchlaufen, usage_counters zeigt
  reales +29/-29-Delta (beide Meldungen real angewendet)

Pruefungen siehe mail/docs/ARC-01-PRUEFPROTOKOLL.md

mail/internal/storage/service_test.go

@@ -0,0 +1,134 @@
+package storage
+
+import (
+	"context"
+	"errors"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+type fakeUsageReporter struct {
+	reports []int64
+}
+
+func (f *fakeUsageReporter) Report(_ context.Context, _, metric string, delta int64) error {
+	if metric != UsageMetric {
+		return errors.New("unerwartete metrik: " + metric)
+	}
+	f.reports = append(f.reports, delta)
+	return nil
+}
+
+func newTestService(t *testing.T) (*Service, *fakeUsageReporter) {
+	t.Helper()
+	driver := NewLocalDriver(t.TempDir())
+	usage := &fakeUsageReporter{}
+	return NewService(driver, usage, "acme"), usage
+}
+
+// TestPut_ReadBackIsByteIdentical ist die geforderte Pflichtprüfung 1:
+// ein geschriebenes Objekt liefert beim Lesen byteidentischen Inhalt.
+func TestPut_ReadBackIsByteIdentical(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+	key := ObjectKey("msg-1", 0)
+	content := "vollständig identischer Inhalt äöü"
+
+	checksum, err := svc.Put(ctx, key, strings.NewReader(content), int64(len(content)), "text/plain")
+	if err != nil {
+		t.Fatalf("put: %v", err)
+	}
+	if checksum == "" {
+		t.Fatal("erwartet nicht-leere prüfsumme")
+	}
+
+	got, err := svc.GetVerified(ctx, key)
+	if err != nil {
+		t.Fatalf("getverified: %v", err)
+	}
+	if string(got) != content {
+		t.Fatalf("nicht byteidentisch: got %q, want %q", got, content)
+	}
+}
+
+// TestGetVerified_DetectsTamperedObject ist die geforderte
+// Pflichtprüfung 2: ein absichtlich beschädigtes Objekt wird bei
+// Prüfsummenvergleich erkannt.
+func TestGetVerified_DetectsTamperedObject(t *testing.T) {
+	dir := t.TempDir()
+	driver := NewLocalDriver(dir)
+	usage := &fakeUsageReporter{}
+	svc := NewService(driver, usage, "acme")
+	ctx := context.Background()
+	key := ObjectKey("msg-tamper", 0)
+
+	if _, err := svc.Put(ctx, key, strings.NewReader("originaler inhalt"), 17, "text/plain"); err != nil {
+		t.Fatalf("put: %v", err)
+	}
+
+	// Objekt DIREKT am Dateisystem manipulieren — umgeht Service
+	// vollständig, simuliert externe Beschädigung/Manipulation.
+	full := driver.path(key)
+	if err := os.WriteFile(full, []byte("MANIPULIERTER INHALT"), 0o644); err != nil {
+		t.Fatalf("manipulation schreiben: %v", err)
+	}
+
+	_, err := svc.GetVerified(ctx, key)
+	if !errors.Is(err, ErrChecksumMismatch) {
+		t.Fatalf("erwartet ErrChecksumMismatch bei manipuliertem objekt, habe: %v", err)
+	}
+}
+
+// TestPut_ManySmallObjectsAcceptableLatency ist die geforderte
+// Pflichtprüfung 3: Lasttest mit vielen kleinen Objekten bestätigt
+// akzeptable Latenz.
+func TestPut_ManySmallObjectsAcceptableLatency(t *testing.T) {
+	svc, _ := newTestService(t)
+	ctx := context.Background()
+
+	const count = 500
+	start := time.Now()
+	for i := 0; i < count; i++ {
+		key := ObjectKey("msg-load", i)
+		if _, err := svc.Put(ctx, key, strings.NewReader("kleiner anhang inhalt"), 21, "text/plain"); err != nil {
+			t.Fatalf("put #%d: %v", i, err)
+		}
+	}
+	elapsed := time.Since(start)
+	avgPerObject := elapsed / count
+
+	// Großzügige Grenze (10ms/Objekt inkl. Schreiben+Sidecar+Rücklese-
+	// Verifikation) — Ziel ist der Nachweis, dass keine quadratische
+	// oder anderweitig unverhältnismäßige Verschlechterung auftritt,
+	// nicht ein knallhartes Performance-SLA.
+	if avgPerObject > 10*time.Millisecond {
+		t.Fatalf("erwartet akzeptable latenz (<10ms/objekt), habe %v/objekt (gesamt %v für %d objekte)", avgPerObject, elapsed, count)
+	}
+	t.Logf("Lasttest: %d Objekte in %v (%v/Objekt)", count, elapsed, avgPerObject)
+}
+
+// TestPut_ReportsUsageOnWriteAndDelete ist die geforderte
+// Pflichtprüfung 4: Melde-Aufruf an Core LIC-05 bei Schreib- und
+// Löschvorgang nachweislich ausgelöst, mit korrekter Größenangabe.
+func TestPut_ReportsUsageOnWriteAndDelete(t *testing.T) {
+	svc, usage := newTestService(t)
+	ctx := context.Background()
+	key := ObjectKey("msg-usage", 0)
+	content := "zwölf bytes!"
+
+	if _, err := svc.Put(ctx, key, strings.NewReader(content), int64(len(content)), "text/plain"); err != nil {
+		t.Fatalf("put: %v", err)
+	}
+	if len(usage.reports) != 1 || usage.reports[0] != int64(len(content)) {
+		t.Fatalf("erwartet genau eine positive meldung mit größe %d, habe: %v", len(content), usage.reports)
+	}
+
+	if err := svc.Delete(ctx, key, int64(len(content))); err != nil {
+		t.Fatalf("delete: %v", err)
+	}
+	if len(usage.reports) != 2 || usage.reports[1] != -int64(len(content)) {
+		t.Fatalf("erwartet zusätzliche negative meldung mit -%d, habe: %v", len(content), usage.reports)
+	}
+}