webhooker/internal/delivery/target_database_archive.go

package delivery

import (
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"sync"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	"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 = "rwc"

	// 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 = "rw"
)

var (
	// errArchiveMissingWebhookID is returned when an event to
	// archive has no webhook id to key its archive file on.
	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 webhook was deleted, or its last database
	// target was removed) 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",
	)
)

// 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
// per-webhook archive database 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 per-webhook 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
	// per-webhook 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

	// sweepOwned marks a registry entry that the idle sweep
	// created because no writer was cached for the webhook. The
	// sweep removes such an entry again when it is done, so a
	// sweep can never leave — or resurrect — a registry entry
	// for a webhook 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 {
	dbURL := fmt.Sprintf("file:%s?mode=%s", w.path, mode)

	sqlDB, err := sql.Open("sqlite", dbURL)
	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
}

// evict closes the writer's handle and marks it unusable. It is
// called when the writer leaves the registry, either because the
// webhook was deleted or because its last database target was
// removed. 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
}