webhooker/internal/archive/archive.go

// Package archive manages the SQLite archive files the database target
// writes: how they are named and rotated, the writers that open, write,
// close and reopen them, pruning the rows past a target's expiry,
// renaming a target's files and moving them back when that fails, the
// gzipped JSON export, the idle sweep, and the locks that keep a rename
// apart from archive writes, from other edits and from exports.
package archive

import (
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	"go.uber.org/fx"
	"sneak.berlin/go/webhooker/internal/database"
)

// archiveNameMaxLen is how many characters of a webhook or target
// name an archive file name keeps.
const archiveNameMaxLen = 40

// Archives is how the handlers keep the database targets' archive
// files in step with the configuration. Deleting a webhook or a
// target releases the cached archive writers, whose open file
// handles would otherwise outlive them; renaming one renames the
// archive files, which are named for the webhook and the target
// (see FileName). RenameLock returns the lock every rename runs under
// (see Writers.RenameLock).
//
// It is an interface so that the handlers' tests can stand in for
// Writers.
//
// Neither eviction deletes an archive file. Both are idempotent
// and are no-ops for a webhook or target with no writer.
type Archives interface {
	RenameLock() *sync.Mutex
	EvictWebhook(webhookID string)
	EvictTarget(targetID string)
	Rename(targetID, webhookName, targetName string) error
}

// WritersParams holds the fx dependencies for Writers.
type WritersParams struct {
	fx.In

	Database  *database.Database
	DBManager *database.WebhookDBManager
	Logger    *slog.Logger
}

// Writers holds the archive writer of each database target, and the
// lock every rename of archive files runs under. The database target
// writes an event through it, the handlers rename and evict through
// it, the archive sweeper prunes through it, and the delivery engine
// evicts every writer when it stops. See archiveWriter for the
// close/reopen, auto-recreate, and expiry semantics.
type Writers struct {
	db    *database.Database
	dbMgr *database.WebhookDBManager
	log   *slog.Logger

	// writers holds one archive writer per database target, keyed
	// by target ID.
	mu      sync.Mutex
	writers map[string]*archiveWriter

	// renameMu makes the webhook edit, the target edit and target
	// creation run one at a time, each held from loading the stored
	// names through the archive rename, the save and any move back.
	// Interleaved, one could rename an archive between another's
	// rename and save, leaving the file named for one edit and the
	// stored names from the other. An archive download holds it while
	// it reads the stored names and lists the files they give, and
	// again for each file while it finds the file under the names
	// stored then and opens it. RenameLock returns it.
	renameMu sync.Mutex
}

// NewWriters builds the archive writers. It opens no archive file:
// each target's writer is made on its first use.
func NewWriters(params WritersParams) *Writers {
	return &Writers{
		db:    params.Database,
		dbMgr: params.DBManager,
		log:   params.Logger,
	}
}

// RenameLock implements Archives. It returns the lock every rename of
// archive files runs under: see renameMu for who holds it and for how
// long. Archive writes and Rename itself do not take it.
func (ws *Writers) RenameLock() *sync.Mutex {
	return &ws.renameMu
}

// FileName returns the file name of a database target's
// archive: archive-WEBHOOKNAME-TARGETNAME-TARGETID.db, with both
// names passed through archiveNamePart. The target ID keeps the
// name unique when two targets' names come out the same.
func FileName(webhookName, targetName, targetID string) string {
	return "archive-" + archiveNamePart(webhookName) + "-" +
		archiveNamePart(targetName) + "-" + targetID + ".db"
}

// archiveNamePart makes a webhook or target name safe to put in a
// file name. It is lowercased; ASCII letters and digits are kept,
// every other run of characters becomes a single "-", and no "-" is
// left at either end. It is cut to archiveNameMaxLen characters, and
// a name with nothing left is "unnamed".
func archiveNamePart(name string) string {
	var b strings.Builder

	dash := false

	for _, r := range strings.ToLower(name) {
		if (r < 'a' || r > 'z') && (r < '0' || r > '9') {
			dash = b.Len() > 0

			continue
		}

		if dash {
			b.WriteByte('-')

			dash = false
		}

		b.WriteRune(r)
	}

	part := b.String()
	if len(part) > archiveNameMaxLen {
		part = strings.TrimRight(part[:archiveNameMaxLen], "-")
	}

	if part == "" {
		return "unnamed"
	}

	return part
}

// Write writes the full event of a delivery to a database target as
// a row into the target's archive database, honouring the target's
// ArchiveExpiry and ArchiveRotation. With rotation, the event goes to
// the file for the period of its receive time.
func (ws *Writers) Write(d *database.Delivery) error {
	webhookID := d.Event.WebhookID
	if webhookID == "" {
		return errArchiveMissingWebhookID
	}

	// An event whose stored row was gone before its delivery ran has
	// no receive time (see delivery's Engine.hydrateEvent), and goes
	// to the file for now.
	receivedAt := d.Event.CreatedAt
	if receivedAt.IsZero() {
		receivedAt = time.Now()
	}

	w, err := ws.writerFor(d.TargetID)
	if err != nil {
		return err
	}

	row := archivedEvent{
		EventID:      d.Event.ID,
		WebhookID:    webhookID,
		EntrypointID: d.Event.EntrypointID,
		Method:       d.Event.Method,
		RawQuery:     d.Event.RawQuery,
		Headers:      d.Event.Headers,
		Body:         d.Event.Body,
		ContentType:  d.Event.ContentType,
	}

	return w.write(
		row, d.Target.ArchiveExpiry,
		archivePeriod(d.Target.ArchiveRotation, receivedAt),
	)
}

// Rename implements Archives. It renames every one of a database
// target's archive files to FileName(webhookName, targetName,
// targetID), each keeping the period in its name. It goes through the
// target's writer, so the move holds the lock that the target's
// archive writes and the idle sweep take, and later writes use the
// new name. It never replaces a file: if one already has a new name,
// the error is ErrNameTaken.
//
// The writer is created if there is none, and it stays cached. The
// handlers rename before they save the new name, so until the save
// the main database still has the old one; a delivery in that window
// must find this writer rather than build one from the old name.
func (ws *Writers) Rename(
	targetID, webhookName, targetName string,
) error {
	w, err := ws.writerFor(targetID)
	if err != nil {
		return err
	}

	return w.rename(FileName(webhookName, targetName, targetID))
}

// EvictTarget implements Archives. It drops a database target's
// archive writer from the registry and closes its handle, so a
// deleted target does not leave a writer (and an open archive handle
// within its debounce window) alive for the process lifetime.
//
// The map entry is removed under the registry lock, which is
// then released before the handle is closed under the writer's
// own lock: that ordering keeps the registry available to other
// targets while an in-flight write on this one drains, and
// closing under the writer's lock means eviction can never race
// a write.
//
// Eviction is idempotent and silent for a target with no writer,
// which is the common case: only a database target that has
// received an event or been renamed has one. It never deletes the
// archive file.
func (ws *Writers) EvictTarget(targetID string) {
	ws.mu.Lock()

	w, ok := ws.writers[targetID]
	if ok {
		delete(ws.writers, targetID)
	}

	ws.mu.Unlock()

	if !ok {
		return
	}

	w.evict()

	ws.log.Info(
		"evicted archive writer",
		"target_id", targetID,
		"path", w.path,
	)
}

// EvictWebhook implements Archives. It evicts, exactly as EvictTarget
// does, the writer of every database target of a webhook. The archive
// files themselves are left on disk — they are long-term storage the
// operator owns.
func (ws *Writers) EvictWebhook(webhookID string) {
	ws.mu.Lock()

	var gone []*archiveWriter

	for targetID, w := range ws.writers {
		if w.webhookID == webhookID {
			delete(ws.writers, targetID)

			gone = append(gone, w)
		}
	}

	ws.mu.Unlock()

	for _, w := range gone {
		w.evict()

		ws.log.Info(
			"evicted archive writer",
			"webhook_id", webhookID,
			"path", w.path,
		)
	}
}

// EvictAll evicts every cached archive writer, exactly as EvictTarget
// does for one target. The delivery engine calls it at shutdown, once
// its workers have returned. Closing the last handle on an archive
// moves the contents of its -wal into the .db and removes the
// -wal, so a clean stop leaves each archive as a single file.
func (ws *Writers) EvictAll() {
	ws.mu.Lock()

	writers := ws.writers
	ws.writers = nil

	ws.mu.Unlock()

	for _, w := range writers {
		w.evict()
	}
}

// writerFor returns the archive writer for a database target,
// creating and caching it on first use. Each target has one writer
// so its close/reopen debounce state is shared across concurrent
// deliveries, and so a rename and the idle sweep take the same lock
// as its writes.
func (ws *Writers) writerFor(
	targetID string,
) (*archiveWriter, error) {
	ws.mu.Lock()
	defer ws.mu.Unlock()

	w, ok := ws.writers[targetID]
	if !ok {
		var err error

		w, err = ws.newWriter(targetID)
		if err != nil {
			return nil, err
		}

		if ws.writers == nil {
			ws.writers = make(map[string]*archiveWriter)
		}

		ws.writers[targetID] = w
	}

	// A delivery claims the entry: even if the idle sweep created
	// it moments ago, it now belongs to the registry proper and
	// the sweep must leave it in place when it finishes.
	w.sweepOwned = false

	return w, nil
}

// sweepWriterFor returns the archive writer the idle sweep should
// prune a target's archive through, together with whether the sweep
// itself created the registry entry.
//
// The sweep must route its prune through the registered writer so
// the writer's mutex orders it against concurrent writes, but it
// must never leave a registry entry behind: a sweep that ran
// concurrently with the target's deletion would otherwise
// re-create an entry that nothing will ever evict again, which is
// exactly the leak eviction exists to prevent. An entry the sweep
// creates is therefore marked sweep-owned and handed back to
// releaseSweepWriter when the sweep is done.
func (ws *Writers) sweepWriterFor(
	targetID string,
) (*archiveWriter, bool, error) {
	ws.mu.Lock()
	defer ws.mu.Unlock()

	w, ok := ws.writers[targetID]
	if ok {
		return w, false, nil
	}

	w, err := ws.newWriter(targetID)
	if err != nil {
		return nil, false, err
	}

	if ws.writers == nil {
		ws.writers = make(map[string]*archiveWriter)
	}

	w.sweepOwned = true
	ws.writers[targetID] = w

	return w, true, nil
}

// releaseSweepWriter drops a registry entry that the idle sweep
// created, so a sweep leaves the registry exactly as it found it.
//
// The entry is removed only if it is still the very writer the
// sweep installed and no delivery has claimed it in the meantime
// (writerFor clears sweepOwned when it hands a writer to the
// write path). Both conditions are evaluated under the registry
// lock, so an eviction that raced the sweep — which removes the
// entry outright — simply finds nothing left to do here, and a
// delivery that adopted the writer keeps a registered, evictable
// one.
func (ws *Writers) releaseSweepWriter(
	targetID string, w *archiveWriter,
) {
	ws.mu.Lock()
	defer ws.mu.Unlock()

	cur, ok := ws.writers[targetID]
	if !ok || cur != w || !cur.sweepOwned {
		return
	}

	delete(ws.writers, targetID)
}

// newWriter builds the writer for a database target's archive. Its
// path is the one Path gives for the webhook and the target as
// the main database names them now; from then on only rename changes
// the name the writer uses. It does not touch the archive files.
func (ws *Writers) newWriter(
	targetID string,
) (*archiveWriter, error) {
	if ws.dbMgr == nil {
		return nil, errArchiveNoDataDir
	}

	var target database.Target

	err := ws.db.DB().
		Preload("Webhook").
		First(&target, "id = ?", targetID).Error
	if err != nil {
		return nil, fmt.Errorf(
			"loading database target %s: %w", targetID, err,
		)
	}

	w := newArchiveWriter(
		Path(ws.dbMgr, &target.Webhook, &target),
		ws.log,
	)
	w.webhookID = target.WebhookID

	return w, nil
}

// sweepArchive prunes one database target's archive of rows older
// than expiry, without requiring a write. A missing archive file is
// left missing (see sweepExpired), so a sweep never creates an
// archive for a target that has never received an event.
//
// It also never leaves a registry entry behind: an entry it had
// to create to reach the writer's mutex is released again once
// the prune is done, so a sweep racing a target deletion cannot
// resurrect the writer the eviction just dropped.
func (ws *Writers) sweepArchive(
	targetID string, expiry time.Duration,
) error {
	w, created, err := ws.sweepWriterFor(targetID)
	if err != nil {
		return err
	}

	if created {
		defer ws.releaseSweepWriter(targetID, w)
	}

	return w.sweepExpired(expiry)
}