webhooker/internal/database/event_db_isolation.go

package database

import (
	"fmt"
	"log/slog"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

// omitAssociationsCallback is the name the association guard is
// registered under on a per-webhook database's create and update
// callback chains.
const omitAssociationsCallback = "webhooker:omit_associations"

// omitAssociations makes every create and update issued against a
// per-webhook database skip GORM's automatic association save.
//
// A per-webhook database holds the event tier only, but Delivery
// declares belongs-to Event and Target and the delivery engine fills
// both in memory before writing. Without this guard GORM upserts
// those parent rows here on the delivery and retry write paths,
// copying targets.config, which holds destination URLs and bearer
// credentials, into the file most likely to be backed up or handed
// to someone else. Registering the guard on the connection covers
// every write path, including writes inside a transaction and write
// paths added later. Every event-tier row this file holds is written
// explicitly, so nothing depends on the automatic save.
func omitAssociations(db *gorm.DB) error {
	omit := func(tx *gorm.DB) {
		tx.Statement.Omits = append(
			tx.Statement.Omits, clause.Associations,
		)
	}

	err := db.Callback().Create().
		Before("gorm:save_before_associations").
		Register(omitAssociationsCallback, omit)
	if err != nil {
		return fmt.Errorf(
			"registering create association guard: %w", err,
		)
	}

	err = db.Callback().Update().
		Before("gorm:save_before_associations").
		Register(omitAssociationsCallback, omit)
	if err != nil {
		return fmt.Errorf(
			"registering update association guard: %w", err,
		)
	}

	return nil
}

// eventDBSweptVersion is the PRAGMA user_version purgeTargetRows
// stamps into a per-webhook database once it has removed any leaked
// target rows *and* the VACUUM that removes their bytes has returned.
// Nothing else in the tree uses user_version, so 0 means "not swept
// by this build".
//
// The stamp, not the DELETE, is what records that a file is done. A
// DELETE commits on its own, so a sweep that is interrupted or whose
// VACUUM fails leaves a file whose rows are gone but whose credential
// bytes are still in the free pages -- indistinguishable, by row
// count, from a file that never leaked. Both leave the stamp unset,
// so the next open sweeps again.
const eventDBSweptVersion = 1

// purgeTargetRows deletes target rows that an earlier build's
// association upsert wrote into a per-webhook database, and rewrites
// the file so their bytes are gone with them. AutoMigrate creates a
// targets table in every one of these files because Delivery declares
// a belongs-to Target, but nothing in the event tier may put rows in
// it. The rows it did put there are junk, not history: they carry an
// empty webhook_id, and delivery rows resolve their target against
// the main database, so nothing here refers to them.
//
// The DELETE only unlinks the rows: modernc.org/sqlite leaves
// secure_delete at SQLite's default of off, so the credential bytes
// stay readable in the file's free pages and a backup of a swept file
// would still hand them over. VACUUM rewrites the file without them.
//
// This runs before every migration and is gated on
// eventDBSweptVersion, so a file pays for the rewrite once, on the
// first open that finds it unstamped, and every open after that is a
// PRAGMA read. A file this build created is stamped before its
// targets table exists, so it never vacuums at all. A failure here
// fails the open with the stamp left unset, so the sweep is retried
// rather than skipped -- a webhook whose file cannot be swept stays
// unusable instead of quietly serving from a file that still holds
// recoverable credentials.
func purgeTargetRows(
	db *gorm.DB, log *slog.Logger, webhookID string,
) error {
	var version int

	// Row().Scan, not (*gorm.DB).Scan: see internal/gormlog.
	err := db.Raw("PRAGMA user_version").Row().Scan(&version)
	if err != nil {
		return fmt.Errorf(
			"reading sweep marker of webhook database %s: %w",
			webhookID, err,
		)
	}

	if version >= eventDBSweptVersion {
		return nil
	}

	var purged int64

	if db.Migrator().HasTable("targets") {
		res := db.Exec("DELETE FROM targets")
		if res.Error != nil {
			return fmt.Errorf(
				"purging target rows from webhook database %s: %w",
				webhookID, res.Error,
			)
		}

		purged = res.RowsAffected

		// Unconditional: a zero row count here does not mean there is
		// nothing to remove, only that no *live* row is left. See
		// eventDBSweptVersion.
		err = db.Exec("VACUUM").Error
		if err != nil {
			return fmt.Errorf(
				"purged %d leaked target rows from webhook database "+
					"%s but vacuuming it failed, so the deleted "+
					"target credentials are still recoverable from "+
					"the file; it stays marked unswept and the next "+
					"open retries: %w",
				purged, webhookID, err,
			)
		}
	}

	err = db.Exec(fmt.Sprintf(
		"PRAGMA user_version = %d", eventDBSweptVersion,
	)).Error
	if err != nil {
		return fmt.Errorf(
			"marking webhook database %s swept: %w", webhookID, err,
		)
	}

	if purged > 0 {
		log.Warn(
			"purged leaked target rows from per-webhook database",
			"webhook_id", webhookID,
			"rows", purged,
		)
	}

	return nil
}