webhooker/internal/delivery/target_database_test.go

package delivery_test

import (
	"database/sql"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
	_ "modernc.org/sqlite" // Pure Go SQLite driver.
	"sneak.berlin/go/webhooker/internal/archive"
	"sneak.berlin/go/webhooker/internal/database"
	"sneak.berlin/go/webhooker/internal/database/databasetest"
	"sneak.berlin/go/webhooker/internal/delivery"
	"sneak.berlin/go/webhooker/internal/gormlog"
)

// archiveTestWebhookName is the name of every webhook
// seedDatabaseTarget creates. It is not safe in a file name as it
// stands, so every archive file name made from it is made safe first.
const archiveTestWebhookName = "Sweep Test!"

// archiveEnv bundles the pieces a database target test drives: a main
// configuration database holding webhooks and targets, a delivery
// engine whose archive writers write into the data directory, and that
// directory.
type archiveEnv struct {
	eng     *delivery.Engine
	mainDB  *database.Database
	dataDir string
}

func setupArchiveTest(t *testing.T) *archiveEnv {
	t.Helper()

	dataDir := t.TempDir()

	mainDB, err := database.Open(dataDir, slog.New(slog.DiscardHandler))
	require.NoError(t, err)

	t.Cleanup(func() { _ = mainDB.Close() })

	return &archiveEnv{
		eng: delivery.NewTestEngineWithDB(
			mainDB,
			databasetest.NewWebhookDBManager(t, dataDir),
			archiveTestLogger(),
			&http.Client{Timeout: 5 * time.Second},
			1,
		),
		mainDB:  mainDB,
		dataDir: dataDir,
	}
}

func archiveTestLogger() *slog.Logger {
	return slog.New(slog.NewTextHandler(
		os.Stderr,
		&slog.HandlerOptions{Level: slog.LevelDebug},
	))
}

// archivePath returns where the engine keeps a database target's
// archive file, for the names seedDatabaseTarget gave it.
func (env *archiveEnv) archivePath(tgt *database.Target) string {
	return filepath.Join(
		env.dataDir,
		archive.FileName(archiveTestWebhookName, tgt.Name, tgt.ID),
	)
}

// seedDatabaseTarget creates a webhook with one database target
// carrying the given target config JSON, and returns the target.
func (env *archiveEnv) seedDatabaseTarget(
	t *testing.T, configJSON string,
) *database.Target {
	t.Helper()

	wh := &database.Webhook{
		UserID: uuid.New().String(),
		Name:   archiveTestWebhookName,
	}
	require.NoError(
		t,
		env.mainDB.DB().
			Omit(clause.Associations).
			Create(wh).Error,
	)

	tgt := &database.Target{
		WebhookID:   wh.ID,
		Name:        "Archive",
		Type:        database.TargetTypeDatabase,
		Active:      true,
		Config:      configJSON,
		MaxAttempts: 1,
	}
	require.NoError(
		t,
		env.mainDB.DB().
			Omit(clause.Associations).
			Create(tgt).Error,
	)

	return tgt
}

// seedDatabaseTargetDelivery seeds a pending delivery of an event
// to a database target and returns the in-memory delivery the
// target handler is invoked with.
func seedDatabaseTargetDelivery(
	t *testing.T,
	webhookDB *gorm.DB,
	event database.Event,
	tgt *database.Target,
) *database.Delivery {
	t.Helper()

	dlv := seedDelivery(
		t, webhookDB, event.ID, tgt.ID,
		database.DeliveryStatusPending,
	)

	d := &database.Delivery{
		EventID:  event.ID,
		TargetID: tgt.ID,
		Status:   database.DeliveryStatusPending,
		Event:    event,
		Target: database.Target{
			Name:   tgt.Name,
			Type:   database.TargetTypeDatabase,
			Config: tgt.Config,
		},
	}
	d.ID = dlv.ID

	return d
}

// openArchiveDBForRead opens an archive file read-only so a
// test can inspect the rows the database target archived.
func openArchiveDBForRead(
	t *testing.T, path string,
) *gorm.DB {
	t.Helper()

	sqlDB, err := sql.Open(
		"sqlite",
		fmt.Sprintf("file:%s?mode=ro", path),
	)
	require.NoError(t, err)

	t.Cleanup(func() { _ = sqlDB.Close() })

	gdb, err := gorm.Open(
		sqlite.Dialector{Conn: sqlDB},
		&gorm.Config{Logger: gormlog.New(slog.New(slog.DiscardHandler))},
	)
	require.NoError(t, err)

	return gdb
}

// countArchivedRows counts the rows in an archive file without
// asserting anything, so it is safe to poll from an
// assert.Eventually condition (which runs off the test
// goroutine, where testify assertions must not be used).
func countArchivedRows(path string) (int64, error) {
	sqlDB, err := sql.Open(
		"sqlite", fmt.Sprintf("file:%s?mode=ro", path),
	)
	if err != nil {
		return 0, err
	}

	defer func() { _ = sqlDB.Close() }()

	gdb, err := gorm.Open(
		sqlite.Dialector{Conn: sqlDB},
		&gorm.Config{Logger: gormlog.New(slog.New(slog.DiscardHandler))},
	)
	if err != nil {
		return 0, err
	}

	var count int64

	err = gdb.Table("archived_events").Count(&count).Error
	if err != nil {
		return 0, err
	}

	return count, nil
}

// archivedRow is the columns of an archive row these tests read.
type archivedRow struct {
	EventID   string
	WebhookID string
	Method    string
	RawQuery  string
	Body      string
}

// TestDeliverDatabase_ArchivesEvent verifies that delivering to
// a database target marks the delivery delivered and archives
// the full event into the target's own archive file.
func TestDeliverDatabase_ArchivesEvent(t *testing.T) {
	t.Parallel()

	env := setupArchiveTest(t)
	tgt := env.seedDatabaseTarget(t, "")

	webhookDB := testWebhookDB(t)
	event := seedEvent(t, webhookDB, `{"archived":true}`)
	event.RawQuery = eventQuery
	d := seedDatabaseTargetDelivery(t, webhookDB, event, tgt)

	env.eng.ExportDeliverDatabase(webhookDB, d)

	var updated database.Delivery

	require.NoError(t, webhookDB.First(
		&updated, "id = ?", d.ID,
	).Error)
	assert.Equal(t,
		database.DeliveryStatusDelivered, updated.Status,
		"database target should mark the delivery delivered",
	)

	archivePath := filepath.Join(
		env.dataDir, "archive-sweep-test-archive-"+tgt.ID+".db",
	)
	assert.FileExists(t, archivePath)

	rdb := openArchiveDBForRead(t, archivePath)

	var rows []archivedRow

	require.NoError(t, rdb.Table("archived_events").Find(&rows).Error)
	require.Len(t, rows, 1)
	assert.Equal(t, event.ID, rows[0].EventID)
	assert.Equal(t, event.WebhookID, rows[0].WebhookID)
	assert.Equal(t, event.Method, rows[0].Method)
	assert.Equal(t, eventQuery, rows[0].RawQuery)
	assert.JSONEq(t, `{"archived":true}`, rows[0].Body)
}

// TestDeliverDatabase_ArchiveFailureFailsDelivery verifies that
// an archive error (here: an unparseable expiry in the target
// config) fails the delivery loudly: the attempt is recorded as
// failed with the error and the delivery is marked failed, not
// delivered.
func TestDeliverDatabase_ArchiveFailureFailsDelivery(
	t *testing.T,
) {
	t.Parallel()

	env := setupArchiveTest(t)
	tgt := env.seedDatabaseTarget(t, `{"expiry":"nonsense"}`)

	webhookDB := testWebhookDB(t)
	event := seedEvent(t, webhookDB, `{"archived":false}`)
	d := seedDatabaseTargetDelivery(t, webhookDB, event, tgt)

	env.eng.ExportDeliverDatabase(webhookDB, d)

	var updated database.Delivery

	require.NoError(t, webhookDB.First(
		&updated, "id = ?", d.ID,
	).Error)
	assert.Equal(t,
		database.DeliveryStatusFailed, updated.Status,
		"archive failure must mark the delivery failed",
	)

	var results []database.DeliveryResult

	require.NoError(t, webhookDB.Where(
		"delivery_id = ?", d.ID,
	).Find(&results).Error)
	require.Len(t, results, 1)
	assert.False(t,
		results[0].Success,
		"the attempt must be recorded as failed",
	)
	assert.Contains(t,
		results[0].Error, "nonsense",
		"the archive error must be recorded on the attempt",
	)

	assert.NoFileExists(t,
		env.archivePath(tgt),
		"no archive file should exist for a failed config",
	)
}