webhooker/internal/delivery/target_database_archive.go

package delivery

import (
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"log/slog"
	"os"
	"path/filepath"
	"sync"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	"sneak.berlin/go/webhooker/internal/database"
	"sneak.berlin/go/webhooker/internal/gormlog"
)

// archiveExpiryNever is the expiry sentinel (and default) that
// disables pruning so archived rows are kept forever.
const archiveExpiryNever = "never"

// archiveReopenDebounce bounds how often an archive file is
// closed and reopened. After each write the handle is closed
// and reopened so an operator can move the file away for
// offline archiving, but never more than once per this window.
const archiveReopenDebounce = time.Second

const (
	// archiveModeCreate is the SQLite URI mode used by the write
	// path: open the archive file, creating it if missing, so a
	// first write (or a write after the operator moved the file
	// away) recreates it.
	archiveModeCreate = database.SQLiteModeCreate

	// archiveModeExisting is the SQLite URI mode used by the idle
	// sweep: open read-write but never create. A sweep must never
	// conjure an empty archive file for a webhook that has a
	// database target but has never received an event.
	archiveModeExisting = database.SQLiteModeExisting
)

var (
	// errArchiveMissingWebhookID is returned when an event to
	// archive has no webhook id to record in its archive row.
	errArchiveMissingWebhookID = errors.New(
		"cannot archive event without a webhook id",
	)

	// errArchiveNoDataDir is returned when the database target
	// has no webhook database manager and so cannot locate the
	// data directory for archive files.
	errArchiveNoDataDir = errors.New(
		"database target has no data directory",
	)

	// errArchiveExpiryNotPositive is returned when a
	// user-supplied archive expiry parses as a duration but is
	// zero or negative; "never" is the way to disable pruning.
	errArchiveExpiryNotPositive = errors.New(
		"expiry must be a positive duration or \"never\"",
	)

	// errArchiveWriterEvicted is returned when a writer that has
	// been evicted (its target or its webhook was deleted) is used
	// again. An evicted writer is no longer in the registry, so
	// reopening its file would leak a handle nothing owns.
	errArchiveWriterEvicted = errors.New(
		"archive writer has been evicted",
	)

	// ErrArchiveNameTaken is returned when an archive cannot be
	// renamed because a file already has the new name. That file may
	// be an archive with rows of its own, so it is never replaced.
	ErrArchiveNameTaken = errors.New(
		"a file already has the archive's new name",
	)
)

// databaseTargetConfig is the optional per-target JSON config
// for a database (archive) target.
type databaseTargetConfig struct {
	// Expiry is a Go duration (e.g. "720h") after which
	// archived rows are pruned, or "never" (the default) to
	// keep them forever.
	Expiry string `json:"expiry"`
}

// archivedEvent is one fully captured webhook event stored in a
// database target's archive for long-term retention. It is a
// self-contained copy — independent of the per-webhook event
// database, which may prune events under its own retention.
type archivedEvent struct {
	ID           uint   `gorm:"primaryKey;autoIncrement"`
	EventID      string `gorm:"index"`
	WebhookID    string
	EntrypointID string
	Method       string
	Headers      string
	Body         string
	ContentType  string

	// ArchivedAt is when the row was archived and is the age
	// basis for expiry pruning.
	ArchivedAt time.Time `gorm:"index"`
}

// parseArchiveExpiry reads the optional expiry from a database
// target's config JSON. An empty config, an empty expiry, or
// the literal "never" all mean keep forever, returned as a zero
// duration. Any other value must parse as a positive Go
// duration; a set-but-invalid value (unparseable, zero, or
// negative) is an error rather than a silent default, matching
// ValidateArchiveExpiry at target creation.
func parseArchiveExpiry(
	configJSON string,
) (time.Duration, error) {
	if configJSON == "" {
		return 0, nil
	}

	var cfg databaseTargetConfig

	err := json.Unmarshal([]byte(configJSON), &cfg)
	if err != nil {
		return 0, fmt.Errorf(
			"parsing database target config: %w", err,
		)
	}

	if cfg.Expiry == "" || cfg.Expiry == archiveExpiryNever {
		return 0, nil
	}

	dur, err := time.ParseDuration(cfg.Expiry)
	if err != nil {
		return 0, fmt.Errorf(
			"parsing archive expiry %q: %w", cfg.Expiry, err,
		)
	}

	if dur <= 0 {
		return 0, fmt.Errorf(
			"%w: %q", errArchiveExpiryNotPositive, cfg.Expiry,
		)
	}

	return dur, nil
}

// ValidateArchiveExpiry checks a user-supplied archive expiry
// for a database target at configuration time. Valid values are
// empty, "never" (both meaning keep forever), or a positive Go
// duration such as "720h". Anything else is an error, so a bad
// expiry is rejected when the target is created rather than
// failing every subsequent delivery.
func ValidateArchiveExpiry(expiry string) error {
	if expiry == "" || expiry == archiveExpiryNever {
		return nil
	}

	dur, err := time.ParseDuration(expiry)
	if err != nil {
		return fmt.Errorf(
			"expiry must be %q or a Go duration "+
				"such as \"720h\": %w",
			archiveExpiryNever, err,
		)
	}

	if dur <= 0 {
		return fmt.Errorf(
			"%w: %q", errArchiveExpiryNotPositive, expiry,
		)
	}

	return nil
}

// archiveWriter owns one database target's archive SQLite file.
// It serialises writes, and after each write closes and reopens
// the file (debounced to at most once per debounce window) so
// an operator can move the file away for offline archiving. The
// next write recreates a moved or removed file, because the
// file is opened create-if-missing and its schema is migrated
// on every open.
type archiveWriter struct {
	mu         sync.Mutex
	path       string
	log        *slog.Logger
	debounce   time.Duration
	db         *gorm.DB
	lastReopen time.Time
	reopens    int

	// evicted marks a writer that has been removed from the
	// registry. Its handle is closed and it must never open the
	// file again: nothing holds it any more, so a reopen would
	// leak the handle for the process lifetime.
	evicted bool

	// webhookID is the webhook the archive's target belongs to,
	// so deleting the webhook can find its writers. It is set
	// when the writer is created and never changes.
	webhookID string

	// sweepOwned marks a registry entry that the idle sweep
	// created because no writer was cached for the target. The
	// sweep removes such an entry again when it is done, so a
	// sweep can never leave — or resurrect — a registry entry
	// for a target that has been deleted. A delivery that adopts
	// the writer clears the flag, handing the entry to the
	// registry proper.
	//
	// Unlike every other field here it is guarded by
	// databaseTarget.mu, not by this writer's mu: it describes the
	// registry entry rather than the file.
	sweepOwned bool
}

// newArchiveWriter builds an archiveWriter for a file path with
// the default reopen debounce.
func newArchiveWriter(
	path string, log *slog.Logger,
) *archiveWriter {
	return &archiveWriter{
		path:     path,
		log:      log,
		debounce: archiveReopenDebounce,
	}
}

// write appends the event as a row, then applies the debounced
// close/reopen. It recreates the archive file if it was moved
// or removed since the last open. A positive expiry prunes rows
// older than it on each (re)open.
func (w *archiveWriter) write(
	row archivedEvent, expiry time.Duration,
) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	if w.evicted {
		return fmt.Errorf(
			"%w: %s", errArchiveWriterEvicted, w.path,
		)
	}

	if w.db == nil || !fileExists(w.path) {
		err := w.reopen(expiry)
		if err != nil {
			return err
		}
	}

	row.ArchivedAt = time.Now()

	err := w.db.Create(&row).Error
	if err != nil {
		return fmt.Errorf(
			"archiving event to %s: %w", w.path, err,
		)
	}

	if time.Since(w.lastReopen) >= w.debounce {
		return w.reopen(expiry)
	}

	return nil
}

// open opens (creating if missing) the archive file, migrates
// its schema, records the reopen time, and prunes expired rows
// when expiry is positive.
func (w *archiveWriter) open(expiry time.Duration) error {
	return w.openMode(archiveModeCreate, expiry)
}

// openMode opens the archive file with the given SQLite URI
// mode, migrates its schema, records the reopen time, and
// prunes expired rows when expiry is positive. The write path
// passes archiveModeCreate so a missing file is recreated; the
// idle sweep passes archiveModeExisting so a missing file is an
// error rather than a newly conjured empty archive.
func (w *archiveWriter) openMode(
	mode string, expiry time.Duration,
) error {
	// Opened through database.OpenSQLite so an archive file carries
	// the same WAL journaling, busy timeout, immediate-transaction
	// locking, and pool bounds as every other database file. See
	// internal/database/sqlite_open.go.
	sqlDB, err := database.OpenSQLite(w.path, mode)
	if err != nil {
		return fmt.Errorf(
			"opening archive database %s: %w", w.path, err,
		)
	}

	gdb, err := gorm.Open(
		sqlite.Dialector{Conn: sqlDB}, &gorm.Config{
			// Never leave this at GORM's default. See
			// internal/gormlog.
			Logger: gormlog.New(w.log),
		},
	)
	if err != nil {
		_ = sqlDB.Close()

		return fmt.Errorf(
			"connecting to archive database %s: %w",
			w.path, err,
		)
	}

	err = gdb.AutoMigrate(&archivedEvent{})
	if err != nil {
		_ = sqlDB.Close()

		return fmt.Errorf(
			"migrating archive database %s: %w", w.path, err,
		)
	}

	w.db = gdb
	w.lastReopen = time.Now()
	w.reopens++

	if expiry > 0 {
		w.prune(expiry)
	}

	return nil
}

// reopen closes any open handle and opens the file afresh. The
// fresh open recreates the file if it was moved away.
func (w *archiveWriter) reopen(expiry time.Duration) error {
	w.close()

	return w.open(expiry)
}

// close closes the underlying handle, if any.
func (w *archiveWriter) close() {
	if w.db == nil {
		return
	}

	sqlDB, err := w.db.DB()
	if err == nil {
		_ = sqlDB.Close()
	}

	w.db = nil
}

// sweepExpired prunes an archive that may have gone idle, with
// no write to trigger the usual on-reopen prune. It takes the
// writer's own mutex for the whole operation, so a sweep is
// ordered against concurrent writes rather than reaching around
// them to the file.
//
// It never creates the archive file: a missing file is skipped,
// and the reopen uses archiveModeExisting so SQLite itself
// refuses to create one if the file disappears between the
// check and the open.
//
// The archive is left CLOSED afterwards. An idle archive holding
// no handle is what keeps the operator's move-the-file-away
// workflow working; the next write reopens (and recreates) the
// file as it always has.
func (w *archiveWriter) sweepExpired(expiry time.Duration) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	if w.evicted {
		return fmt.Errorf(
			"%w: %s", errArchiveWriterEvicted, w.path,
		)
	}

	if !fileExists(w.path) {
		return nil
	}

	// Drop any live handle first so the prune runs against a
	// freshly opened file, matching the write path's semantics.
	w.close()

	err := w.openMode(archiveModeExisting, expiry)
	if err != nil {
		return err
	}

	w.close()

	return nil
}

// rename gives the archive file a new name in the same directory,
// and the writer uses the file under that name from now on. The
// handle is closed first, which folds the -wal into the .db; any
// -wal or -shm still beside the file (left by a crash) is moved with
// it, because SQLite finds them by name. A missing file is not an
// error: the operator may have moved it away, and the next write
// creates it under the new name.
//
// If a file already has the new name, nothing is moved and the
// error is ErrArchiveNameTaken. If one file fails to move, those
// already moved are moved back before the error is returned, so the
// archive is never split across two names.
func (w *archiveWriter) rename(name string) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	if w.evicted {
		return fmt.Errorf(
			"%w: %s", errArchiveWriterEvicted, w.path,
		)
	}

	path := filepath.Join(filepath.Dir(w.path), name)
	if path == w.path {
		return nil
	}

	suffixes := []string{"", "-wal", "-shm"}

	for _, suffix := range suffixes {
		if fileExists(path + suffix) {
			return fmt.Errorf(
				"%w: %s", ErrArchiveNameTaken, name+suffix,
			)
		}
	}

	w.close()

	for i, suffix := range suffixes {
		err := os.Rename(w.path+suffix, path+suffix)
		if err == nil || errors.Is(err, fs.ErrNotExist) {
			continue
		}

		for _, moved := range suffixes[:i] {
			backErr := os.Rename(path+moved, w.path+moved)
			if backErr != nil && !errors.Is(backErr, fs.ErrNotExist) {
				w.log.Error(
					"failed to move archive file back",
					"from", path+moved,
					"to", w.path+moved,
					"error", backErr,
				)
			}
		}

		return fmt.Errorf(
			"renaming archive %s to %s: %w", w.path+suffix, path+suffix, err,
		)
	}

	w.path = path

	return nil
}

// evict closes the writer's handle and marks it unusable. It is
// called when the writer leaves the registry, because its target
// or its webhook was deleted, or at shutdown. The archive FILE is
// deliberately left on disk: it is long-term storage an operator
// may still want.
func (w *archiveWriter) evict() {
	w.mu.Lock()
	defer w.mu.Unlock()

	w.evicted = true

	w.close()
}

// prune deletes archived rows older than expiry, measured from
// each row's archived time. It runs on every (re)open, so a
// steadily written archive is swept by its own write traffic. An
// archive that goes idle receives no further reopens, which is
// why ArchiveSweeper exists to drive sweepExpired on a timer.
// Failures are logged, not fatal: a prune error must not stop
// archiving.
func (w *archiveWriter) prune(expiry time.Duration) {
	cutoff := time.Now().Add(-expiry)

	res := w.db.Where("archived_at < ?", cutoff).
		Delete(&archivedEvent{})
	if res.Error != nil {
		w.log.Error(
			"failed to prune expired archive rows",
			"path", w.path,
			"error", res.Error,
		)

		return
	}

	if res.RowsAffected > 0 {
		w.log.Info(
			"pruned expired archive rows",
			"path", w.path,
			"rows_deleted", res.RowsAffected,
		)
	}
}

// fileExists reports whether a path currently exists.
func fileExists(path string) bool {
	_, err := os.Stat(path)

	return err == nil
}