syncova-backup/packages/repository/session.go

package repository

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"sync"
	"time"
)

// journalState ist der Zustand einer Schreibsession.
type journalState string

const (
	// journalStateOpen bedeutet: die Session schreibt noch.
	journalStateOpen journalState = "open"
	// journalStateCommitting bedeutet: der Abschluss läuft.
	//
	// Findet sich beim Start ein Journal in diesem Zustand, brach der Vorgang
	// mitten im Commit ab und muss geprüft werden.
	journalStateCommitting journalState = "committing"
)

// sessionJournal hält den Fortschritt einer Schreibsession fest.
//
// Das Journal macht einen Abbruch nachvollziehbar: nach einem Absturz lässt
// sich erkennen, welche Sessions offen blieben und aufgeräumt werden müssen
// (PROMPT.md §82).
type sessionJournal struct {
	// SessionID ist die Kennung der Schreibsession.
	SessionID string `json:"session_id"`
	// BackupID ist das Backup, das geschrieben wird.
	BackupID string `json:"backup_id"`
	// State ist der Zustand der Session.
	State journalState `json:"state"`
	// StartedAt ist der Beginn der Session in UTC.
	StartedAt time.Time `json:"started_at"`
	// UpdatedAt ist der Zeitpunkt der letzten Fortschreibung in UTC.
	UpdatedAt time.Time `json:"updated_at"`
	// Statistics sind die bisher erfassten Kennzahlen.
	Statistics SessionStatistics `json:"statistics"`
	// ProcessID ist der schreibende Prozess.
	ProcessID int `json:"process_id"`
}

// localWriter ist die Schreibsession eines lokalen Repositorys.
type localWriter struct {
	// repository ist das Ziel-Repository.
	repository *LocalRepository
	// sessionID ist die Kennung dieser Session.
	sessionID string
	// backupID ist das zu schreibende Backup.
	backupID string
	// startedAt ist der Beginn der Session in UTC.
	startedAt time.Time
	// statisticsMutex schützt die Kennzahlen gegen gleichzeitige Schreiber.
	statisticsMutex sync.Mutex
	// statistics sind die laufenden Kennzahlen.
	statistics SessionStatistics
	// isClosed meldet, ob die Session bereits abgeschlossen oder abgebrochen ist.
	isClosed bool
	// logger protokolliert den Verlauf der Session.
	logger *slog.Logger
}

// BeginBackup öffnet eine Schreibsession (Schritt 1 des Commit-Protokolls).
func (localRepository *LocalRepository) BeginBackup(beginContext context.Context, backupID string) (Writer, error) {
	if validationError := validateBackupIdentifier(backupID); validationError != nil {
		return nil, validationError
	}

	// Ein bereits vorhandenes Backup darf nicht überschrieben werden: das wäre
	// stiller Datenverlust (PROMPT.md §141).
	if _, statError := os.Stat(localRepository.manifestPath(backupID)); statError == nil {
		return nil, fmt.Errorf("zu der kennung %s existiert bereits ein backup", backupID)
	}

	// Ohne Schreibsperre könnten zwei Vorgänge denselben Katalog überschreiben.
	if localRepository.lockHandle == nil {
		return nil, fmt.Errorf("das repository wurde nur lesend geöffnet und kann nicht beschrieben werden")
	}

	sessionID := backupID
	startedAt := localRepository.timeSource().UTC()

	if directoryError := ensureDirectory(localRepository.stagingPath(sessionID)); directoryError != nil {
		return nil, directoryError
	}

	sessionWriter := &localWriter{
		repository: localRepository,
		sessionID:  sessionID,
		backupID:   backupID,
		startedAt:  startedAt,
		logger: localRepository.logger.With(
			slog.String("backup_id", backupID),
			slog.String("session_id", sessionID),
		),
	}

	if journalError := sessionWriter.writeJournal(journalStateOpen); journalError != nil {
		return nil, journalError
	}

	sessionWriter.logger.Info("schreibsession geöffnet")

	return sessionWriter, nil
}

// WriteChunk legt einen Datenblock ab (Schritt 2 des Commit-Protokolls).
func (sessionWriter *localWriter) WriteChunk(writeContext context.Context, chunkData []byte) (ChunkReference, bool, error) {
	sessionWriter.statisticsMutex.Lock()
	if sessionWriter.isClosed {
		sessionWriter.statisticsMutex.Unlock()
		return ChunkReference{}, false, ErrSessionClosed
	}
	sessionWriter.statisticsMutex.Unlock()

	// Ein abgebrochener Vorgang soll nicht weiterschreiben.
	if contextError := writeContext.Err(); contextError != nil {
		return ChunkReference{}, false, contextError
	}

	chunkIdentifier, wasNew, writeError := sessionWriter.repository.writeChunk(writeContext, chunkData)
	if writeError != nil {
		return ChunkReference{}, false, writeError
	}

	chunkLength := int64(len(chunkData))

	sessionWriter.statisticsMutex.Lock()
	sessionWriter.statistics.LogicalBytes += chunkLength

	if wasNew {
		sessionWriter.statistics.ChunksWritten++
		sessionWriter.statistics.StoredBytes += chunkLength
	} else {
		// Ein bereits vorhandener Chunk spart genau seine Länge ein.
		sessionWriter.statistics.ChunksDeduplicated++
		sessionWriter.statistics.DeduplicatedBytes += chunkLength
	}
	sessionWriter.statisticsMutex.Unlock()

	return ChunkReference{
		Identifier:    chunkIdentifier,
		LogicalLength: chunkLength,
		StoredLength:  chunkLength,
	}, wasNew, nil
}

// WriteTransformedChunk legt einen bereits umgewandelten Block ab und zählt ihn.
//
// Die Backup Engine komprimiert und verschlüsselt selbst; sie übergibt die
// fertige Form. Gezählt wird beides: die Klartextlänge als verarbeitete
// Datenmenge und die abgelegte Länge als tatsächlich geschriebene.
//
// Der Unterschied zwischen beiden ist der Gewinn aus Kompression — und der
// Grund, warum die Klartextlänge als Parameter kommt: Aus dem Geheimtext lässt
// sie sich nicht zurückrechnen.
func (sessionWriter *localWriter) WriteTransformedChunk(writeContext context.Context,
	plaintextIdentifier string, storedData []byte, plaintextLength int64) (string, bool, error) {
	sessionWriter.statisticsMutex.Lock()
	if sessionWriter.isClosed {
		sessionWriter.statisticsMutex.Unlock()

		return "", false, ErrSessionClosed
	}
	sessionWriter.statisticsMutex.Unlock()

	if contextError := writeContext.Err(); contextError != nil {
		return "", false, contextError
	}

	storedDigest, wasNew, writeError := sessionWriter.repository.WriteTransformedChunk(
		writeContext, plaintextIdentifier, storedData)
	if writeError != nil {
		return "", false, writeError
	}

	storedLength := int64(len(storedData))

	sessionWriter.statisticsMutex.Lock()
	sessionWriter.statistics.LogicalBytes += plaintextLength

	if wasNew {
		sessionWriter.statistics.ChunksWritten++
		sessionWriter.statistics.StoredBytes += storedLength
	} else {
		// Ein bereits vorhandener Block spart genau seine abgelegte Länge.
		sessionWriter.statistics.ChunksDeduplicated++
		sessionWriter.statistics.DeduplicatedBytes += storedLength
	}
	sessionWriter.statisticsMutex.Unlock()

	return storedDigest, wasNew, nil
}

// NoteDeduplicatedChunk vermerkt einen bereits vorhandenen Block.
//
// Gezaehlt wird die **Klartextlaenge**, nicht die abgelegte: Die Frage, die
// diese Zahl beantwortet, lautet „wie viel musste nicht abgelegt werden" — und
// die bemisst sich an den Ursprungsdaten.
func (sessionWriter *localWriter) NoteDeduplicatedChunk(plaintextLength int64) {
	sessionWriter.statisticsMutex.Lock()
	defer sessionWriter.statisticsMutex.Unlock()

	if sessionWriter.isClosed {
		return
	}

	sessionWriter.statistics.LogicalBytes += plaintextLength
	sessionWriter.statistics.ChunksDeduplicated++
	sessionWriter.statistics.DeduplicatedBytes += plaintextLength
}

// Commit schließt das Backup ab (Schritte 3 bis 7 des Commit-Protokolls).
//
// Der Ablauf folgt SYNCOVA_ARCHITECTURE.md §10:
//
//  3. Manifest schreiben
//  4. Manifest prüfen
//  5. Abschlussvermerk atomar setzen
//  6. Katalog aktualisieren
//  7. Erfolgszustand veröffentlichen
//
// Bricht der Vorgang vor Schritt 5 ab, bleibt kein sichtbares Backup zurück.
func (sessionWriter *localWriter) Commit(commitContext context.Context, manifest *Manifest) error {
	sessionWriter.statisticsMutex.Lock()
	if sessionWriter.isClosed {
		sessionWriter.statisticsMutex.Unlock()
		return ErrSessionClosed
	}
	currentStatistics := sessionWriter.statistics
	sessionWriter.statisticsMutex.Unlock()

	if manifest == nil {
		return errors.New("ohne manifest kann kein backup abgeschlossen werden")
	}

	// Das Journal hält fest, dass der Abschluss begonnen hat.
	if journalError := sessionWriter.writeJournal(journalStateCommitting); journalError != nil {
		return journalError
	}

	// Die Angaben des Aufrufers werden mit den tatsächlich erfassten Werten
	// überschrieben: erfundene Kennzahlen wären eine Falschaussage (PROMPT.md §138).
	manifest.ManifestVersion = ManifestVersion
	manifest.BackupID = sessionWriter.backupID
	manifest.StartedAt = sessionWriter.startedAt
	manifest.CompletedAt = sessionWriter.repository.timeSource().UTC()
	manifest.Statistics = currentStatistics

	if manifest.ChainID == "" {
		// Ohne Kettenangabe bildet das Backup seine eigene Kette.
		manifest.ChainID = sessionWriter.backupID
	}

	if manifest.BackupType == "" {
		manifest.BackupType = BackupTypeFull
	}

	if manifest.ConsistencyLevel == "" {
		// Ohne ausdrückliche Zusicherung gilt die schwächste Stufe. Eine höhere
		// anzunehmen hiesse, eine nicht erbrachte Eigenschaft zu behaupten.
		manifest.ConsistencyLevel = ConsistencyCrash
	}

	// Jedes von dieser Programmversion geschriebene Manifest ist vollständig:
	// Auch eine Zusatzsicherung trägt die Blockverweise aller unveränderten
	// Objekte (Phase 6). Die Zusicherung wird hier festgehalten, damit die
	// Aufbewahrung ein Elternbackup löschen darf, ohne das Kind unbrauchbar zu
	// machen.
	manifest.SelfContainedRestore = true

	// Ein gehärtetes Repository setzt den Aufbewahrungsschutz selbst durch.
	//
	// Die Frist steht im Descriptor des Repositorys, nicht im Aufruf: Wer ein
	// Backup schreibt, soll die geltende Frist nicht unterlaufen können, indem
	// er einfach eine kürzere mitgibt.
	if sessionWriter.repository.descriptor.Immutable {
		earliestAllowedEnd := manifest.CompletedAt.Add(sessionWriter.repository.descriptor.RetentionPeriod())

		if manifest.ImmutableUntil == nil || manifest.ImmutableUntil.Before(earliestAllowedEnd) {
			manifest.ImmutableUntil = &earliestAllowedEnd
		}
	}

	// Schritt 3 und 4: Prüfsumme bilden und sofort gegenprüfen.
	if sealError := sealManifest(manifest); sealError != nil {
		return sealError
	}

	if verifyError := VerifyManifest(manifest); verifyError != nil {
		return fmt.Errorf("das erzeugte manifest hat die eigene prüfung nicht bestanden: %w", verifyError)
	}

	encodedManifest, encodeError := encodeManifest(manifest)
	if encodeError != nil {
		return encodeError
	}

	// Schritt 5: Das atomare Umbenennen macht das Backup in einem Zug sichtbar.
	// Vorher existiert es für Leser nicht, nachher vollständig.
	manifestPermissions := dataFilePermissions
	if sessionWriter.repository.descriptor.Immutable {
		manifestPermissions = immutableFilePermissions
	}

	manifestFilePath := sessionWriter.repository.manifestPath(sessionWriter.backupID)

	if writeError := writeFileAtomically(manifestFilePath, encodedManifest, manifestPermissions); writeError != nil {
		return fmt.Errorf("das manifest konnte nicht abgelegt werden: %w", writeError)
	}

	// Der Löschschutz wird **nach** dem Sichtbarwerden gesetzt: Vorher liegt das
	// Manifest unter temporärem Namen, und ein unveränderliches Ziel liesse sich
	// nicht mehr per rename ersetzen — der Commit selbst schlüge fehl.
	//
	// Gelingt das Kennzeichen nicht, bleibt es beim Software-Schutz. Ein
	// geschriebenes Backup ohne technischen Löschschutz ist besser als gar
	// keines; verschwiegen wird es nicht, sondern über die gemessene
	// Durchsetzungsstufe ausgewiesen.
	if sessionWriter.repository.descriptor.Immutable {
		if protectError := protectManifestFile(manifestFilePath); protectError != nil {
			sessionWriter.logger.Warn("das manifest konnte nicht gegen löschung geschützt werden; "+
				"es gilt nur der schutz durch diese software",
				slog.String("grund", protectError.Error()))
		}
	}

	// Schritt 6: Der Katalog folgt dem Manifest. Schlägt er fehl, ist das
	// Backup dennoch vollständig und wiederherstellbar - der Katalog lässt
	// sich jederzeit neu aufbauen.
	if catalogError := sessionWriter.repository.appendToCatalog(manifest); catalogError != nil {
		sessionWriter.logger.Error("der katalog konnte nicht aktualisiert werden; das backup ist dennoch vollständig",
			slog.String("error", catalogError.Error()))
	}

	sessionWriter.finish()

	// Schritt 7: Erst jetzt gilt das Backup als erfolgreich.
	sessionWriter.logger.Info("backup abgeschlossen",
		slog.Int64("chunks_written", currentStatistics.ChunksWritten),
		slog.Int64("chunks_deduplicated", currentStatistics.ChunksDeduplicated),
		slog.Int64("logical_bytes", currentStatistics.LogicalBytes),
		slog.Int64("stored_bytes", currentStatistics.StoredBytes),
	)

	return nil
}

// Abort bricht die Session ab und räumt auf.
//
// Bereits geschriebene Chunks bleiben liegen: sie sind inhaltsadressiert und
// werden von einem späteren Lauf wiederverwendet. Ein Manifest entsteht nicht,
// weshalb kein unvollständiges Backup sichtbar wird.
func (sessionWriter *localWriter) Abort(abortContext context.Context) error {
	sessionWriter.statisticsMutex.Lock()
	if sessionWriter.isClosed {
		sessionWriter.statisticsMutex.Unlock()
		return nil
	}
	sessionWriter.statisticsMutex.Unlock()

	sessionWriter.logger.Warn("schreibsession abgebrochen",
		slog.Int64("chunks_written", sessionWriter.statistics.ChunksWritten))

	sessionWriter.finish()

	return nil
}

// Statistics liefert die bisher erfassten Kennzahlen.
func (sessionWriter *localWriter) Statistics() SessionStatistics {
	sessionWriter.statisticsMutex.Lock()
	defer sessionWriter.statisticsMutex.Unlock()

	return sessionWriter.statistics
}

// finish räumt Journal und Arbeitsverzeichnis der Session weg.
func (sessionWriter *localWriter) finish() {
	sessionWriter.statisticsMutex.Lock()
	sessionWriter.isClosed = true
	sessionWriter.statisticsMutex.Unlock()

	// Aufräumfehler werden protokolliert, aber nicht weitergereicht: das Backup
	// selbst ist zu diesem Zeitpunkt bereits abgeschlossen.
	if removeError := os.Remove(sessionWriter.repository.journalPath(sessionWriter.sessionID)); removeError != nil && !errors.Is(removeError, os.ErrNotExist) {
		sessionWriter.logger.Warn("das journal konnte nicht entfernt werden",
			slog.String("error", removeError.Error()))
	}

	if removeError := os.RemoveAll(sessionWriter.repository.stagingPath(sessionWriter.sessionID)); removeError != nil {
		sessionWriter.logger.Warn("das arbeitsverzeichnis konnte nicht entfernt werden",
			slog.String("error", removeError.Error()))
	}
}

// writeJournal schreibt den aktuellen Zustand der Session fort.
func (sessionWriter *localWriter) writeJournal(currentState journalState) error {
	sessionWriter.statisticsMutex.Lock()
	currentStatistics := sessionWriter.statistics
	sessionWriter.statisticsMutex.Unlock()

	journalEntry := sessionJournal{
		SessionID:  sessionWriter.sessionID,
		BackupID:   sessionWriter.backupID,
		State:      currentState,
		StartedAt:  sessionWriter.startedAt,
		UpdatedAt:  sessionWriter.repository.timeSource().UTC(),
		Statistics: currentStatistics,
		ProcessID:  os.Getpid(),
	}

	encodedJournal, marshalError := json.MarshalIndent(journalEntry, "", "  ")
	if marshalError != nil {
		return fmt.Errorf("das journal konnte nicht erzeugt werden: %w", marshalError)
	}

	if writeError := writeFileAtomically(sessionWriter.repository.journalPath(sessionWriter.sessionID),
		append(encodedJournal, '\n'), dataFilePermissions); writeError != nil {
		return fmt.Errorf("das journal konnte nicht geschrieben werden: %w", writeError)
	}

	return nil
}

// StaleSession beschreibt eine nach einem Abbruch zurückgebliebene Session.
type StaleSession struct {
	// SessionID ist die Kennung der Session.
	SessionID string `json:"session_id"`
	// BackupID ist das betroffene Backup.
	BackupID string `json:"backup_id"`
	// State ist der zuletzt vermerkte Zustand.
	State string `json:"state"`
	// StartedAt ist der Beginn der Session in UTC.
	StartedAt time.Time `json:"started_at"`
	// HasManifest meldet, ob trotz offener Session ein Manifest vorliegt.
	//
	// Trifft das zu, wurde der Abschluss vollzogen und nur das Aufräumen
	// unterbrochen — das Backup ist verwendbar.
	HasManifest bool `json:"has_manifest"`
}

// FindStaleSessions sucht Sessions, die ein früherer Lauf offen gelassen hat.
//
// Der Aufruf gehört an den Start des Dienstes: ein Abbruch darf nicht
// stillschweigend Reste hinterlassen (PROMPT.md §140).
func (localRepository *LocalRepository) FindStaleSessions(scanContext context.Context) ([]StaleSession, error) {
	journalDirectory := localRepository.rootPath + string(os.PathSeparator) + directoryJournals

	directoryEntries, readError := os.ReadDir(journalDirectory)
	if readError != nil {
		if errors.Is(readError, os.ErrNotExist) {
			return nil, nil
		}

		return nil, fmt.Errorf("das journalverzeichnis konnte nicht gelesen werden: %w", readError)
	}

	staleSessions := make([]StaleSession, 0)

	for _, directoryEntry := range directoryEntries {
		if directoryEntry.IsDir() || !hasSuffix(directoryEntry.Name(), journalExtension) {
			continue
		}

		rawJournal, journalReadError := os.ReadFile(journalDirectory + string(os.PathSeparator) + directoryEntry.Name())
		if journalReadError != nil {
			continue
		}

		var journalEntry sessionJournal
		if unmarshalError := json.Unmarshal(rawJournal, &journalEntry); unmarshalError != nil {
			continue
		}

		_, manifestStatError := os.Stat(localRepository.manifestPath(journalEntry.BackupID))

		staleSessions = append(staleSessions, StaleSession{
			SessionID:   journalEntry.SessionID,
			BackupID:    journalEntry.BackupID,
			State:       string(journalEntry.State),
			StartedAt:   journalEntry.StartedAt,
			HasManifest: manifestStatError == nil,
		})
	}

	return staleSessions, nil
}

// CleanupStaleSession räumt eine zurückgebliebene Session weg.
//
// Ein bereits geschriebenes Manifest bleibt unangetastet: der Abschluss war
// vollzogen, nur das Aufräumen wurde unterbrochen.
func (localRepository *LocalRepository) CleanupStaleSession(cleanupContext context.Context, sessionID string) error {
	if validationError := validateBackupIdentifier(sessionID); validationError != nil {
		return validationError
	}

	if removeError := os.Remove(localRepository.journalPath(sessionID)); removeError != nil && !errors.Is(removeError, os.ErrNotExist) {
		return fmt.Errorf("das journal konnte nicht entfernt werden: %w", removeError)
	}

	if removeError := os.RemoveAll(localRepository.stagingPath(sessionID)); removeError != nil {
		return fmt.Errorf("das arbeitsverzeichnis konnte nicht entfernt werden: %w", removeError)
	}

	localRepository.logger.Info("zurückgebliebene schreibsession aufgeräumt",
		slog.String("session_id", sessionID))

	return nil
}