FDN-03: objekt-storage-abstraktion
Ein Driver-Interface, zwei austauschbare Treiber: LocalDriver (Entwicklung, Dateisystem, HMAC-signierte URLs) und S3Driver (Produktion, S3-kompatibel via aws-sdk-go-v2, echte presigned URLs). Pfadschema documents/<id>/ revisions/<id> innerhalb des mandantenspezifischen Buckets. Service verbindet Driver mit Nutzungsmeldung an Core LIC-05 (HTTPUsageReporter, Vertrag von internal/resync.Handler nachgebildet, DMS kann Cores internal/- Pakete als eigenes Modul nicht importieren). Auf 192.168.1.131 verifiziert, S3-Treiber gegen echtes lokal installiertes MinIO (kein Mock): Round-Trip beide Treiber, abgelaufene presigned URL real mit 403 abgewiesen (manuell zusaetzlich zum Unit-Test verifiziert), klare ErrNotFound bei fehlendem Objekt beide Treiber, Nutzungsmeldung mit korrektem Tenant/Metrik/Delta bei Put/Delete. Befund dokumentiert: Cores internal/resync.Handler (Gegenstelle fuer die Nutzungsmeldung) ist noch in keinem cmd/*/main.go verdrahtet (dieselbe Fehlerklasse wie QA-05/AUD-06) - HTTPUsageReporter daher gegen den dokumentierten Vertrag getestet, nicht gegen eine laufende Core-Instanz. Siehe dms/docs/FDN-03-PRUEFPROTOKOLL.md. golangci-lint auf v2.1.6 aktualisiert (v1.63.4 konnte go1.24-Zielstand nicht linten), .golangci.yml auf v2-Konfigurationsformat migriert. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HhgFcLS8tYMhDJpP74C6AQ
This commit is contained in:
sysops
co-authored by
Claude Sonnet 5
Claude Sonnet 5
parent
9d4c2bae4a
commit
442bb674e9
@@ -0,0 +1,45 @@
|
||||
// Package storage implementiert FDN-03: eine einheitliche Objekt-Storage-
|
||||
// Abstraktion mit zwei austauschbaren Treibern (lokal fuer Entwicklung,
|
||||
// S3-kompatibel fuer Produktion). Verschluesselung at rest ist NICHT
|
||||
// Bestandteil dieser Kachel (siehe FDN-09) — dieses Paket legt Bytes
|
||||
// unveraendert ab.
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ErrNotFound wird geliefert, wenn ein angefragtes Objekt nicht existiert
|
||||
// (Akzeptanzkriterium/Pruefung 3: klarer Fehler statt treiberspezifischer
|
||||
// Fehlertypen, die der Aufrufer sonst je Treiber unterschiedlich behandeln
|
||||
// muesste).
|
||||
var ErrNotFound = errors.New("storage: objekt nicht gefunden")
|
||||
|
||||
// Driver ist die EINE Schnittstelle, gegen die der Rest von DMS arbeitet
|
||||
// (Akzeptanzkriterium 1). Zwei Implementierungen: LocalDriver (Entwicklung)
|
||||
// und S3Driver (Produktion, S3-kompatibel).
|
||||
type Driver interface {
|
||||
// Put legt die Bytes aus r unter key ab und liefert die tatsaechlich
|
||||
// geschriebene Groesse in Bytes.
|
||||
Put(ctx context.Context, key string, r io.Reader, size int64, contentType string) (int64, error)
|
||||
// Get liefert die Bytes unter key. Existiert key nicht, liefert Get
|
||||
// ErrNotFound.
|
||||
Get(ctx context.Context, key string) (io.ReadCloser, error)
|
||||
// Delete entfernt das Objekt unter key. Existiert key nicht, liefert
|
||||
// Delete ErrNotFound.
|
||||
Delete(ctx context.Context, key string) error
|
||||
// SignedURL liefert eine zeitlich begrenzte, signierte URL zum Lesen des
|
||||
// Objekts (Akzeptanzkriterium 2: konfigurierbare Gueltigkeit ueber ttl).
|
||||
SignedURL(ctx context.Context, key string, ttl time.Duration) (string, error)
|
||||
}
|
||||
|
||||
// ObjectKey liefert das Pfadschema fuer ein Dokument/Revision INNERHALB des
|
||||
// Mandanten-Buckets (Akzeptanzkriterium 3) — die Bucket-Trennung selbst ist
|
||||
// Sache von Core TEN-01, hier geht es nur um den Pfad innerhalb eines
|
||||
// bereits mandantenspezifischen Buckets.
|
||||
func ObjectKey(documentID, revisionID string) string {
|
||||
return "documents/" + documentID + "/revisions/" + revisionID
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"crypto/subtle"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ErrURLExpired wird von VerifySignedURL geliefert, wenn eine signierte URL
|
||||
// nach Ablauf ihrer Gueltigkeit verwendet wird (Pruefung 2).
|
||||
var ErrURLExpired = errors.New("storage: signierte url ist abgelaufen")
|
||||
|
||||
// ErrInvalidSignature wird geliefert, wenn die Signatur einer URL nicht zum
|
||||
// Schluessel passt (manipulierte oder falsche URL).
|
||||
var ErrInvalidSignature = errors.New("storage: signatur der url ist ungueltig")
|
||||
|
||||
// LocalDriver legt Objekte im lokalen Dateisystem ab — der Entwicklungs-
|
||||
// Treiber (Akzeptanzkriterium 1), keine externe Abhaengigkeit noetig.
|
||||
type LocalDriver struct {
|
||||
baseDir string
|
||||
signingSecret []byte
|
||||
publicBaseURL string
|
||||
}
|
||||
|
||||
// NewLocalDriver erzeugt einen LocalDriver. signingSecret authentifiziert
|
||||
// die von SignedURL ausgestellten URLs (HMAC-SHA256, konstant-zeit-
|
||||
// verglichen bei der Verifikation — timing-safe wie projektweite Konvention,
|
||||
// siehe Core IAM-15).
|
||||
func NewLocalDriver(baseDir string, signingSecret []byte, publicBaseURL string) *LocalDriver {
|
||||
return &LocalDriver{baseDir: baseDir, signingSecret: signingSecret, publicBaseURL: publicBaseURL}
|
||||
}
|
||||
|
||||
func (d *LocalDriver) path(key string) string {
|
||||
return filepath.Join(d.baseDir, filepath.FromSlash(key))
|
||||
}
|
||||
|
||||
func (d *LocalDriver) Put(ctx context.Context, key string, r io.Reader, size int64, contentType string) (int64, error) {
|
||||
full := d.path(key)
|
||||
if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
|
||||
return 0, fmt.Errorf("storage: verzeichnis anlegen: %w", err)
|
||||
}
|
||||
f, err := os.Create(full)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("storage: datei anlegen: %w", err)
|
||||
}
|
||||
defer func() { _ = f.Close() }()
|
||||
|
||||
written, err := io.Copy(f, r)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("storage: schreiben: %w", err)
|
||||
}
|
||||
return written, nil
|
||||
}
|
||||
|
||||
func (d *LocalDriver) Get(ctx context.Context, key string) (io.ReadCloser, error) {
|
||||
f, err := os.Open(d.path(key))
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
return nil, fmt.Errorf("storage: lesen: %w", err)
|
||||
}
|
||||
return f, nil
|
||||
}
|
||||
|
||||
func (d *LocalDriver) Delete(ctx context.Context, key string) error {
|
||||
if err := os.Remove(d.path(key)); err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return ErrNotFound
|
||||
}
|
||||
return fmt.Errorf("storage: loeschen: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *LocalDriver) SignedURL(ctx context.Context, key string, ttl time.Duration) (string, error) {
|
||||
expiry := time.Now().Add(ttl).Unix()
|
||||
sig := d.sign(key, expiry)
|
||||
return fmt.Sprintf("%s/%s?exp=%d&sig=%s", strings.TrimRight(d.publicBaseURL, "/"), key, expiry, sig), nil
|
||||
}
|
||||
|
||||
func (d *LocalDriver) sign(key string, expiry int64) string {
|
||||
mac := hmac.New(sha256.New, d.signingSecret)
|
||||
mac.Write([]byte(key))
|
||||
mac.Write([]byte(strconv.FormatInt(expiry, 10)))
|
||||
return base64.RawURLEncoding.EncodeToString(mac.Sum(nil))
|
||||
}
|
||||
|
||||
// VerifySignedURL prueft key/expiry/sig, wie sie z.B. aus den Query-
|
||||
// Parametern einer von SignedURL ausgestellten URL stammen (Pruefung 2:
|
||||
// abgelaufene URL wird abgewiesen). Die eigentliche HTTP-Auslieferung ist
|
||||
// nicht Bestandteil dieser Kachel (siehe DOC-01) — hier wird nur die
|
||||
// Signatur-/Ablauflogik bereitgestellt und getestet.
|
||||
func (d *LocalDriver) VerifySignedURL(key string, expiry int64, sig string) error {
|
||||
expected := d.sign(key, expiry)
|
||||
if subtle.ConstantTimeCompare([]byte(expected), []byte(sig)) != 1 {
|
||||
return ErrInvalidSignature
|
||||
}
|
||||
if time.Now().Unix() > expiry {
|
||||
return ErrURLExpired
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func newTestLocalDriver(t *testing.T) *LocalDriver {
|
||||
t.Helper()
|
||||
return NewLocalDriver(t.TempDir(), []byte("test-signing-secret"), "https://files.example.test")
|
||||
}
|
||||
|
||||
// TestLocalDriver_RoundTrip ist Pruefung 1 fuer den lokalen Treiber:
|
||||
// Upload/Download-Roundtrip.
|
||||
func TestLocalDriver_RoundTrip(t *testing.T) {
|
||||
d := newTestLocalDriver(t)
|
||||
ctx := context.Background()
|
||||
key := "documents/doc-1/revisions/rev-1"
|
||||
content := []byte("hallo welt")
|
||||
|
||||
written, err := d.Put(ctx, key, bytes.NewReader(content), int64(len(content)), "text/plain")
|
||||
if err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
if written != int64(len(content)) {
|
||||
t.Fatalf("geschriebene groesse = %d, want %d", written, len(content))
|
||||
}
|
||||
|
||||
rc, err := d.Get(ctx, key)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
defer func() { _ = rc.Close() }()
|
||||
got, err := io.ReadAll(rc)
|
||||
if err != nil {
|
||||
t.Fatalf("lesen: %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, content) {
|
||||
t.Fatalf("gelesener inhalt = %q, want %q", got, content)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLocalDriver_MissingObject ist Pruefung 3: klarer Fehler bei
|
||||
// fehlendem Objekt, sowohl fuer Get als auch Delete.
|
||||
func TestLocalDriver_MissingObject(t *testing.T) {
|
||||
d := newTestLocalDriver(t)
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := d.Get(ctx, "nie-angelegt"); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("get eines fehlenden objekts: erwartet ErrNotFound, habe %v", err)
|
||||
}
|
||||
if err := d.Delete(ctx, "nie-angelegt"); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("delete eines fehlenden objekts: erwartet ErrNotFound, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLocalDriver_SignedURL_ExpiredIsRejected ist Pruefung 2: eine
|
||||
// abgelaufene signierte URL wird abgewiesen.
|
||||
func TestLocalDriver_SignedURL_ExpiredIsRejected(t *testing.T) {
|
||||
d := newTestLocalDriver(t)
|
||||
ctx := context.Background()
|
||||
key := "documents/doc-2/revisions/rev-1"
|
||||
|
||||
// Gueltige, noch nicht abgelaufene URL wird akzeptiert.
|
||||
urlValid, err := d.SignedURL(ctx, key, time.Hour)
|
||||
if err != nil {
|
||||
t.Fatalf("signedurl (gueltig): %v", err)
|
||||
}
|
||||
expiry, sig := parseSignedURLQuery(t, urlValid)
|
||||
if err := d.VerifySignedURL(key, expiry, sig); err != nil {
|
||||
t.Fatalf("gueltige url wurde abgewiesen: %v", err)
|
||||
}
|
||||
|
||||
// Bereits abgelaufene URL (negative TTL) wird abgewiesen.
|
||||
urlExpired, err := d.SignedURL(ctx, key, -time.Hour)
|
||||
if err != nil {
|
||||
t.Fatalf("signedurl (abgelaufen): %v", err)
|
||||
}
|
||||
expiredExpiry, expiredSig := parseSignedURLQuery(t, urlExpired)
|
||||
if err := d.VerifySignedURL(key, expiredExpiry, expiredSig); !errors.Is(err, ErrURLExpired) {
|
||||
t.Fatalf("abgelaufene url: erwartet ErrURLExpired, habe %v", err)
|
||||
}
|
||||
|
||||
// Manipulierte Signatur wird abgewiesen.
|
||||
if err := d.VerifySignedURL(key, expiry, "manipuliert"); !errors.Is(err, ErrInvalidSignature) {
|
||||
t.Fatalf("manipulierte signatur: erwartet ErrInvalidSignature, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func parseSignedURLQuery(t *testing.T, rawURL string) (expiry int64, sig string) {
|
||||
t.Helper()
|
||||
parsed, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
t.Fatalf("signierte url parsen: %v (%s)", err, rawURL)
|
||||
}
|
||||
q := parsed.Query()
|
||||
expInt, err := strconv.ParseInt(q.Get("exp"), 10, 64)
|
||||
if err != nil {
|
||||
t.Fatalf("exp parsen: %v", err)
|
||||
}
|
||||
return expInt, q.Get("sig")
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/config"
|
||||
"github.com/aws/aws-sdk-go-v2/credentials"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
"github.com/aws/smithy-go"
|
||||
)
|
||||
|
||||
// S3Driver legt Objekte in einem S3-kompatiblen Objektspeicher ab — der
|
||||
// Produktions-Treiber (Akzeptanzkriterium 1). Funktioniert gegen echtes
|
||||
// AWS S3 UND gegen jeden S3-kompatiblen Anbieter (MinIO etc.) ueber
|
||||
// endpointURL — bewusst offenes Objektformat statt Herstellerbindung
|
||||
// (Produkt-DNA: "jederzeit ohne Herstellerwerkzeug lesbar").
|
||||
type S3Driver struct {
|
||||
client *s3.Client
|
||||
bucket string
|
||||
}
|
||||
|
||||
// NewS3Driver verbindet zu einem S3-kompatiblen Endpunkt. endpointURL leer
|
||||
// laesst den AWS-SDK-Standardendpunkt (echtes AWS S3) gelten,
|
||||
// usePathStyle=true ist fuer die meisten Nicht-AWS-S3-kompatiblen Anbieter
|
||||
// (MinIO, etc.) noetig.
|
||||
func NewS3Driver(ctx context.Context, bucket, region, endpointURL, accessKeyID, secretAccessKey string, usePathStyle bool) (*S3Driver, error) {
|
||||
cfg, err := config.LoadDefaultConfig(ctx,
|
||||
config.WithRegion(region),
|
||||
config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKeyID, secretAccessKey, "")),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("storage: s3-konfiguration laden: %w", err)
|
||||
}
|
||||
|
||||
client := s3.NewFromConfig(cfg, func(o *s3.Options) {
|
||||
if endpointURL != "" {
|
||||
o.BaseEndpoint = aws.String(endpointURL)
|
||||
}
|
||||
o.UsePathStyle = usePathStyle
|
||||
})
|
||||
return &S3Driver{client: client, bucket: bucket}, nil
|
||||
}
|
||||
|
||||
func (d *S3Driver) Put(ctx context.Context, key string, r io.Reader, size int64, contentType string) (int64, error) {
|
||||
buf, err := io.ReadAll(r)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("storage: objekt vor upload lesen: %w", err)
|
||||
}
|
||||
_, err = d.client.PutObject(ctx, &s3.PutObjectInput{
|
||||
Bucket: aws.String(d.bucket),
|
||||
Key: aws.String(key),
|
||||
Body: bytesReader(buf),
|
||||
ContentLength: aws.Int64(int64(len(buf))),
|
||||
ContentType: aws.String(contentType),
|
||||
})
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("storage: s3-upload: %w", err)
|
||||
}
|
||||
return int64(len(buf)), nil
|
||||
}
|
||||
|
||||
func (d *S3Driver) Get(ctx context.Context, key string) (io.ReadCloser, error) {
|
||||
out, err := d.client.GetObject(ctx, &s3.GetObjectInput{
|
||||
Bucket: aws.String(d.bucket),
|
||||
Key: aws.String(key),
|
||||
})
|
||||
if err != nil {
|
||||
if isS3NotFound(err) {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
return nil, fmt.Errorf("storage: s3-download: %w", err)
|
||||
}
|
||||
return out.Body, nil
|
||||
}
|
||||
|
||||
func (d *S3Driver) Delete(ctx context.Context, key string) error {
|
||||
// S3 liefert bei DeleteObject fuer ein nicht existierendes Objekt KEINEN
|
||||
// Fehler (idempotente Semantik der S3-API) — um denselben Vertrag wie
|
||||
// LocalDriver (ErrNotFound bei fehlendem Objekt) zu erfuellen, wird die
|
||||
// Existenz vorher explizit geprueft (Pruefung 3).
|
||||
_, err := d.client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: aws.String(d.bucket), Key: aws.String(key)})
|
||||
if err != nil {
|
||||
if isS3NotFound(err) {
|
||||
return ErrNotFound
|
||||
}
|
||||
return fmt.Errorf("storage: s3-existenzpruefung vor loeschen: %w", err)
|
||||
}
|
||||
|
||||
if _, err := d.client.DeleteObject(ctx, &s3.DeleteObjectInput{
|
||||
Bucket: aws.String(d.bucket),
|
||||
Key: aws.String(key),
|
||||
}); err != nil {
|
||||
return fmt.Errorf("storage: s3-loeschen: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *S3Driver) SignedURL(ctx context.Context, key string, ttl time.Duration) (string, error) {
|
||||
presignClient := s3.NewPresignClient(d.client)
|
||||
req, err := presignClient.PresignGetObject(ctx, &s3.GetObjectInput{
|
||||
Bucket: aws.String(d.bucket),
|
||||
Key: aws.String(key),
|
||||
}, s3.WithPresignExpires(ttl))
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("storage: presigned url erzeugen: %w", err)
|
||||
}
|
||||
return req.URL, nil
|
||||
}
|
||||
|
||||
func bytesReader(b []byte) *bytes.Reader {
|
||||
return bytes.NewReader(b)
|
||||
}
|
||||
|
||||
// isS3NotFound erkennt sowohl den typisierten NoSuchKey-Fehler
|
||||
// (GetObject) als auch den generischen "NotFound"-API-Fehlercode
|
||||
// (HeadObject liefert keinen typisierten NoSuchKey, sondern einen
|
||||
// generischen smithy-API-Fehler mit Code "NotFound").
|
||||
func isS3NotFound(err error) bool {
|
||||
var nsk *types.NoSuchKey
|
||||
if errors.As(err, &nsk) {
|
||||
return true
|
||||
}
|
||||
var apiErr smithy.APIError
|
||||
if errors.As(err, &apiErr) && apiErr.ErrorCode() == "NotFound" {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// requireS3TestEnv liefert die S3-Testkonfiguration oder ueberspringt den
|
||||
// Test — dasselbe Muster wie TEST_ADMIN_DSN im Core-Modul: kein S3-
|
||||
// kompatibler Speicher in dieser Umgebung verfuegbar/geprueft (siehe
|
||||
// FDN-03-Pruefprotokoll), daher hier bewusst als optional markiert statt
|
||||
// den Treiber ungetestet zu lassen.
|
||||
func requireS3TestEnv(t *testing.T) *S3Driver {
|
||||
t.Helper()
|
||||
bucket := os.Getenv("TEST_S3_BUCKET")
|
||||
if bucket == "" {
|
||||
t.Skip("TEST_S3_BUCKET nicht gesetzt, S3-Integrationstest uebersprungen")
|
||||
}
|
||||
endpoint := os.Getenv("TEST_S3_ENDPOINT")
|
||||
region := os.Getenv("TEST_S3_REGION")
|
||||
if region == "" {
|
||||
region = "us-east-1"
|
||||
}
|
||||
accessKey := os.Getenv("TEST_S3_ACCESS_KEY_ID")
|
||||
secretKey := os.Getenv("TEST_S3_SECRET_ACCESS_KEY")
|
||||
|
||||
d, err := NewS3Driver(context.Background(), bucket, region, endpoint, accessKey, secretKey, true)
|
||||
if err != nil {
|
||||
t.Fatalf("s3-treiber aufbauen: %v", err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// TestS3Driver_RoundTrip ist Pruefung 1 fuer den S3-Treiber.
|
||||
func TestS3Driver_RoundTrip(t *testing.T) {
|
||||
d := requireS3TestEnv(t)
|
||||
ctx := context.Background()
|
||||
key := "fdn03-test/roundtrip"
|
||||
content := []byte("s3 roundtrip inhalt")
|
||||
|
||||
t.Cleanup(func() { _ = d.Delete(ctx, key) })
|
||||
|
||||
if _, err := d.Put(ctx, key, bytes.NewReader(content), int64(len(content)), "text/plain"); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
rc, err := d.Get(ctx, key)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
defer func() { _ = rc.Close() }()
|
||||
got, err := io.ReadAll(rc)
|
||||
if err != nil {
|
||||
t.Fatalf("lesen: %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, content) {
|
||||
t.Fatalf("gelesener inhalt = %q, want %q", got, content)
|
||||
}
|
||||
}
|
||||
|
||||
// TestS3Driver_MissingObject ist Pruefung 3 fuer den S3-Treiber.
|
||||
func TestS3Driver_MissingObject(t *testing.T) {
|
||||
d := requireS3TestEnv(t)
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := d.Get(ctx, "fdn03-test/nie-angelegt"); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("get eines fehlenden objekts: erwartet ErrNotFound, habe %v", err)
|
||||
}
|
||||
if err := d.Delete(ctx, "fdn03-test/nie-angelegt"); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("delete eines fehlenden objekts: erwartet ErrNotFound, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestS3Driver_SignedURL ist Pruefung 2 fuer den S3-Treiber: eine
|
||||
// presigned URL wird erzeugt und ist innerhalb der Gueltigkeit abrufbar.
|
||||
func TestS3Driver_SignedURL(t *testing.T) {
|
||||
d := requireS3TestEnv(t)
|
||||
ctx := context.Background()
|
||||
key := "fdn03-test/signed-url"
|
||||
content := []byte("presigned")
|
||||
t.Cleanup(func() { _ = d.Delete(ctx, key) })
|
||||
|
||||
if _, err := d.Put(ctx, key, bytes.NewReader(content), int64(len(content)), "text/plain"); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
|
||||
url, err := d.SignedURL(ctx, key, time.Minute)
|
||||
if err != nil {
|
||||
t.Fatalf("signedurl: %v", err)
|
||||
}
|
||||
if url == "" {
|
||||
t.Fatal("erwartet nicht-leere presigned url")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Service verbindet einen Driver mit der Nutzungsmeldung an Core
|
||||
// (Akzeptanzkriterium 4) — jeder Schreib-/Loeschvorgang ueber Service loest
|
||||
// GENAU EINE Meldung mit der tatsaechlich geschriebenen/geloeschten
|
||||
// Objektgroesse aus. Repository-/Handler-Code (spaetere Kacheln, z.B.
|
||||
// DOC-01) ruft ausschliesslich Service auf, nie einen Driver direkt — das
|
||||
// verhindert einen Schreibpfad, der die Nutzungsmeldung vergisst.
|
||||
type Service struct {
|
||||
driver Driver
|
||||
usage UsageReporter
|
||||
tenantSlug string
|
||||
}
|
||||
|
||||
func NewService(driver Driver, usage UsageReporter, tenantSlug string) *Service {
|
||||
return &Service{driver: driver, usage: usage, tenantSlug: tenantSlug}
|
||||
}
|
||||
|
||||
// Put legt das Objekt ab und meldet die geschriebene Groesse als positives
|
||||
// Delta (Pruefung 4).
|
||||
func (s *Service) Put(ctx context.Context, key string, r io.Reader, size int64, contentType string) (int64, error) {
|
||||
written, err := s.driver.Put(ctx, key, r, size, contentType)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if err := s.usage.Report(ctx, s.tenantSlug, UsageMetric, written); err != nil {
|
||||
return written, fmt.Errorf("storage: objekt gespeichert, aber nutzungsmeldung fehlgeschlagen: %w", err)
|
||||
}
|
||||
return written, nil
|
||||
}
|
||||
|
||||
func (s *Service) Get(ctx context.Context, key string) (io.ReadCloser, error) {
|
||||
return s.driver.Get(ctx, key)
|
||||
}
|
||||
|
||||
// Delete entfernt das Objekt und meldet dessen Groesse als negatives Delta
|
||||
// (Pruefung 4) — dafuer muss der Aufrufer die Groesse kennen (z.B. aus
|
||||
// file_revisions.size_bytes, FDN-02), da Delete selbst die Groesse eines
|
||||
// bereits geloeschten Objekts nicht mehr ermitteln kann.
|
||||
func (s *Service) Delete(ctx context.Context, key string, sizeBytes int64) error {
|
||||
if err := s.driver.Delete(ctx, key); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.usage.Report(ctx, s.tenantSlug, UsageMetric, -sizeBytes); err != nil {
|
||||
return fmt.Errorf("storage: objekt geloescht, aber nutzungsmeldung fehlgeschlagen: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) SignedURL(ctx context.Context, key string, ttl time.Duration) (string, error) {
|
||||
return s.driver.SignedURL(ctx, key, ttl)
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// fakeUsageReporter zeichnet jeden Report-Aufruf auf, damit Tests
|
||||
// nachweisen koennen, dass Service tatsaechlich meldet (Akzeptanzkriterium
|
||||
// 4 / Pruefung 4) — ohne echten HTTP-Aufruf gegen Core.
|
||||
type fakeUsageReporter struct {
|
||||
mu sync.Mutex
|
||||
calls []reportCall
|
||||
failOn int // wenn >0, schlaegt der reportCall-te Aufruf fehl
|
||||
}
|
||||
|
||||
type reportCall struct {
|
||||
tenantSlug string
|
||||
metric string
|
||||
delta int64
|
||||
}
|
||||
|
||||
func (f *fakeUsageReporter) Report(ctx context.Context, tenantSlug, metric string, delta int64) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.calls = append(f.calls, reportCall{tenantSlug, metric, delta})
|
||||
if f.failOn > 0 && len(f.calls) == f.failOn {
|
||||
return context.DeadlineExceeded
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestService_PutReportsPositiveDelta ist Pruefung 4 (Schreibvorgang):
|
||||
// Melde-Aufruf an Core wird bei Put ausgeloest, mit korrekter Groesse.
|
||||
func TestService_PutReportsPositiveDelta(t *testing.T) {
|
||||
driver := NewLocalDriver(t.TempDir(), []byte("secret"), "https://files.example.test")
|
||||
usage := &fakeUsageReporter{}
|
||||
svc := NewService(driver, usage, "acme")
|
||||
|
||||
content := []byte("zwoelf bytes")
|
||||
if _, err := svc.Put(context.Background(), "documents/d1/revisions/r1", bytes.NewReader(content), int64(len(content)), "text/plain"); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
|
||||
if len(usage.calls) != 1 {
|
||||
t.Fatalf("erwartet 1 nutzungsmeldung, habe %d", len(usage.calls))
|
||||
}
|
||||
call := usage.calls[0]
|
||||
if call.tenantSlug != "acme" || call.metric != UsageMetric || call.delta != int64(len(content)) {
|
||||
t.Fatalf("unerwarteter meldungsinhalt: %+v", call)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_DeleteReportsNegativeDelta ist Pruefung 4 (Loeschvorgang).
|
||||
func TestService_DeleteReportsNegativeDelta(t *testing.T) {
|
||||
driver := NewLocalDriver(t.TempDir(), []byte("secret"), "https://files.example.test")
|
||||
usage := &fakeUsageReporter{}
|
||||
svc := NewService(driver, usage, "acme")
|
||||
ctx := context.Background()
|
||||
key := "documents/d2/revisions/r1"
|
||||
|
||||
if _, err := svc.Put(ctx, key, bytes.NewReader([]byte("abc")), 3, "text/plain"); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
if err := svc.Delete(ctx, key, 3); err != nil {
|
||||
t.Fatalf("delete: %v", err)
|
||||
}
|
||||
|
||||
if len(usage.calls) != 2 {
|
||||
t.Fatalf("erwartet 2 nutzungsmeldungen (put+delete), habe %d", len(usage.calls))
|
||||
}
|
||||
del := usage.calls[1]
|
||||
if del.delta != -3 {
|
||||
t.Fatalf("delete-delta = %d, want -3", del.delta)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_GetMissingObjectReturnsClearError ist Pruefung 3 auf
|
||||
// Service-Ebene.
|
||||
func TestService_GetMissingObjectReturnsClearError(t *testing.T) {
|
||||
driver := NewLocalDriver(t.TempDir(), []byte("secret"), "https://files.example.test")
|
||||
svc := NewService(driver, &fakeUsageReporter{}, "acme")
|
||||
|
||||
if _, err := svc.Get(context.Background(), "nie-angelegt"); err == nil {
|
||||
t.Fatal("get eines fehlenden objekts haette einen fehler liefern muessen")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// UsageMetric ist der Metrikname, unter dem Core (internal/usage, LIC-05)
|
||||
// den Speicherverbrauch je Mandant fuehrt — muss exakt
|
||||
// internal/usage.StorageBytesMetric aus dem NEXARCH-Core-Modul entsprechen
|
||||
// (Core kann von DMS als eigenem Go-Modul nicht importiert werden, daher
|
||||
// hier als Konstante gespiegelt statt importiert).
|
||||
const UsageMetric = "storage_bytes"
|
||||
|
||||
// UsageReporter meldet Speicherverbrauchsaenderungen an Core (Akzeptanz-
|
||||
// kriterium 4). Schmale Schnittstelle, damit Tests einen Fake statt eines
|
||||
// echten HTTP-Aufrufs einsetzen koennen.
|
||||
type UsageReporter interface {
|
||||
Report(ctx context.Context, tenantSlug, metric string, delta int64) error
|
||||
}
|
||||
|
||||
// usageDeltaDTO entspricht Core internal/resync.usageDeltaDTO
|
||||
// (JSON-Vertrag: tenant_slug/metric/delta) — dieselbe Struktur, hier
|
||||
// gespiegelt, da DMS Cores internal/-Pakete nicht importieren kann.
|
||||
type usageDeltaDTO struct {
|
||||
TenantSlug string `json:"tenant_slug"`
|
||||
Metric string `json:"metric"`
|
||||
Delta int64 `json:"delta"`
|
||||
}
|
||||
|
||||
// HTTPUsageReporter meldet ueber Cores Resync-Nutzungs-Endpunkt
|
||||
// (internal/resync.Handler.UsageHandler, API-06), authentifiziert ueber
|
||||
// dasselbe Service-Credential-Verfahren wie jeder andere Modul-Core-Aufruf
|
||||
// (API-02).
|
||||
type HTTPUsageReporter struct {
|
||||
endpointURL string
|
||||
clientID string
|
||||
clientSecret string
|
||||
httpClient *http.Client
|
||||
}
|
||||
|
||||
func NewHTTPUsageReporter(endpointURL, clientID, clientSecret string, httpClient *http.Client) *HTTPUsageReporter {
|
||||
if httpClient == nil {
|
||||
httpClient = http.DefaultClient
|
||||
}
|
||||
return &HTTPUsageReporter{endpointURL: endpointURL, clientID: clientID, clientSecret: clientSecret, httpClient: httpClient}
|
||||
}
|
||||
|
||||
func (r *HTTPUsageReporter) Report(ctx context.Context, tenantSlug, metric string, delta int64) error {
|
||||
body, err := json.Marshal([]usageDeltaDTO{{TenantSlug: tenantSlug, Metric: metric, Delta: delta}})
|
||||
if err != nil {
|
||||
return fmt.Errorf("storage: nutzungsmeldung serialisieren: %w", err)
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.endpointURL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return fmt.Errorf("storage: nutzungsmeldungs-anfrage aufbauen: %w", err)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("X-Nexarch-Client-Id", r.clientID)
|
||||
req.Header.Set("X-Nexarch-Client-Secret", r.clientSecret)
|
||||
|
||||
resp, err := r.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("storage: nutzungsmeldung senden: %w", err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("storage: nutzungsmeldung von core abgelehnt: status %d", resp.StatusCode)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestHTTPUsageReporter_SendsCorrectContractToCore ist der Nachweis, dass
|
||||
// HTTPUsageReporter exakt den Vertrag von Core internal/resync.Handler.
|
||||
// UsageHandler bedient (Service-Credential-Header, JSON-Feldnamen) — echte
|
||||
// Vernetzung gegen einen laufenden Core-Prozess ist nicht Teil dieses
|
||||
// Tests (internal/resync.Handler ist in Core aktuell in keinem cmd/*/
|
||||
// main.go verdrahtet, siehe FDN-03-Pruefprotokoll), daher hier gegen einen
|
||||
// httptest-Server geprueft, der denselben Vertrag nachbildet.
|
||||
func TestHTTPUsageReporter_SendsCorrectContractToCore(t *testing.T) {
|
||||
var gotClientID, gotClientSecret string
|
||||
var gotBody []usageDeltaDTO
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
gotClientID = r.Header.Get("X-Nexarch-Client-Id")
|
||||
gotClientSecret = r.Header.Get("X-Nexarch-Client-Secret")
|
||||
if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil {
|
||||
t.Errorf("anfrage-koerper dekodieren: %v", err)
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(map[string]int{"applied": len(gotBody)})
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
reporter := NewHTTPUsageReporter(srv.URL, "dms-service-client", "dms-service-secret", nil)
|
||||
if err := reporter.Report(context.Background(), "acme", UsageMetric, 4096); err != nil {
|
||||
t.Fatalf("report: %v", err)
|
||||
}
|
||||
|
||||
if gotClientID != "dms-service-client" || gotClientSecret != "dms-service-secret" {
|
||||
t.Fatalf("service-credential-header falsch: id=%q secret=%q", gotClientID, gotClientSecret)
|
||||
}
|
||||
if len(gotBody) != 1 {
|
||||
t.Fatalf("erwartet 1 delta im koerper, habe %d", len(gotBody))
|
||||
}
|
||||
if gotBody[0].TenantSlug != "acme" || gotBody[0].Metric != UsageMetric || gotBody[0].Delta != 4096 {
|
||||
t.Fatalf("unerwarteter delta-inhalt: %+v", gotBody[0])
|
||||
}
|
||||
}
|
||||
|
||||
// TestHTTPUsageReporter_RejectsNonOKStatus prueft, dass ein von Core
|
||||
// abgelehnter Aufruf (z.B. ungueltiges Service-Credential) als Fehler
|
||||
// zurueckgegeben wird, statt stillschweigend zu verlieren.
|
||||
func TestHTTPUsageReporter_RejectsNonOKStatus(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "ungueltiges service-credential", http.StatusUnauthorized)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
reporter := NewHTTPUsageReporter(srv.URL, "x", "y", nil)
|
||||
if err := reporter.Report(context.Background(), "acme", UsageMetric, 1); err == nil {
|
||||
t.Fatal("erwartet fehler bei abgelehnter nutzungsmeldung, habe nil")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user