webhooker/internal/database/webhook_db_manager_test.go

package database_test

import (
	"bytes"
	"context"
	"log/slog"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"

	"github.com/google/uuid"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.uber.org/fx/fxtest"
	"gorm.io/gorm"
	"sneak.berlin/go/webhooker/internal/config"
	"sneak.berlin/go/webhooker/internal/database"
	"sneak.berlin/go/webhooker/internal/globals"
	"sneak.berlin/go/webhooker/internal/logger"
)

func setupTestWebhookDBManager(
	t *testing.T,
) (*database.WebhookDBManager, *fxtest.Lifecycle) {
	t.Helper()

	lc := fxtest.NewLifecycle(t)

	g := &globals.Globals{
		Appname: testAppname,
		Version: testVersion,
	}

	l, err := logger.New(
		lc,
		logger.LoggerParams{Globals: g},
	)
	require.NoError(t, err)

	dataDir := filepath.Join(t.TempDir(), "events")

	cfg := &config.Config{
		DataDir: dataDir,
	}

	mgr, err := database.NewWebhookDBManager(
		lc,
		database.WebhookDBManagerParams{
			Config: cfg,
			Logger: l,
		},
	)
	require.NoError(t, err)

	return mgr, lc
}

func TestWebhookDBManager_CreateAndGetDB(t *testing.T) {
	t.Parallel()

	mgr, lc := setupTestWebhookDBManager(t)
	ctx := context.Background()
	require.NoError(t, lc.Start(ctx))

	defer func() { require.NoError(t, lc.Stop(ctx)) }()

	webhookID := uuid.New().String()

	// DB should not exist yet
	assert.False(t, mgr.DBExists(webhookID))

	// Create the DB
	err := mgr.CreateDB(webhookID)
	require.NoError(t, err)

	// DB file should now exist
	assert.True(t, mgr.DBExists(webhookID))

	// Get the DB again (should use cached connection)
	db, err := mgr.GetDB(webhookID)
	require.NoError(t, err)
	require.NotNil(t, db)

	// Verify we can write an event
	event := &database.Event{
		WebhookID:    webhookID,
		EntrypointID: uuid.New().String(),
		Method:       http.MethodPost,
		Headers:      `{"Content-Type":["application/json"]}`,
		Body:         `{"test": true}`,
		ContentType:  testContentType,
	}
	require.NoError(t, db.Create(event).Error)
	assert.NotEmpty(t, event.ID)

	// Verify we can read it back
	var readEvent database.Event

	require.NoError(
		t,
		db.First(&readEvent, "id = ?", event.ID).Error,
	)
	assert.Equal(t, webhookID, readEvent.WebhookID)
	assert.Equal(t, http.MethodPost, readEvent.Method)
	assert.Equal(t, `{"test": true}`, readEvent.Body)
}

// Many callers ask for one webhook's database at the same moment,
// before it is cached. Only one of them may open the file; the others
// must wait for its handle. openDB logs one "opened per-webhook
// database" line per open, and those lines are what is counted.
func TestWebhookDBManager_ConcurrentFirstTouchOpensOnce(t *testing.T) {
	t.Parallel()

	var logs bytes.Buffer

	mgr := database.NewTestWebhookDBManagerWithLogger(
		t.TempDir(),
		slog.New(slog.NewTextHandler(&logs, nil)),
	)

	t.Cleanup(func() { assert.NoError(t, mgr.CloseAll()) })

	webhookID := uuid.New().String()

	const callers = 16

	start := make(chan struct{})
	handles := make([]*gorm.DB, callers)
	errs := make([]error, callers)

	var wg sync.WaitGroup

	for i := range callers {
		wg.Go(func() {
			<-start

			handles[i], errs[i] = mgr.GetDB(webhookID)
		})
	}

	close(start)
	wg.Wait()

	for i := range callers {
		require.NoError(t, errs[i])
		assert.Same(t, handles[0], handles[i])
	}

	assert.Equal(
		t, 1,
		strings.Count(logs.String(), "opened per-webhook database"),
	)
}

func TestWebhookDBManager_DeleteDB(t *testing.T) {
	t.Parallel()

	mgr, lc := setupTestWebhookDBManager(t)
	ctx := context.Background()
	require.NoError(t, lc.Start(ctx))

	defer func() { require.NoError(t, lc.Stop(ctx)) }()

	webhookID := uuid.New().String()

	// Create the DB and write some data
	require.NoError(t, mgr.CreateDB(webhookID))

	db, err := mgr.GetDB(webhookID)
	require.NoError(t, err)

	event := &database.Event{
		WebhookID:    webhookID,
		EntrypointID: uuid.New().String(),
		Method:       http.MethodPost,
		Body:         `{"test": true}`,
		ContentType:  testContentType,
	}
	require.NoError(t, db.Create(event).Error)

	// Under WAL, an open database that has been written to has both
	// sidecars beside it.
	dbPath := mgr.DBPath(webhookID)
	require.FileExists(t, dbPath+"-wal")
	require.FileExists(t, dbPath+"-shm")

	// Delete the DB
	require.NoError(t, mgr.DeleteDB(webhookID))

	// File should no longer exist
	assert.False(t, mgr.DBExists(webhookID))

	// Verify the files are actually gone from disk
	assert.NoFileExists(t, dbPath)
	assert.NoFileExists(t, dbPath+"-wal")
	assert.NoFileExists(t, dbPath+"-shm")
}

// blockRemoval puts a non-empty directory at path, which os.Remove
// cannot remove whoever runs the test, root included.
func blockRemoval(t *testing.T, path string) {
	t.Helper()

	require.NoError(t, os.MkdirAll(filepath.Join(path, "keep"), 0o700))
}

// TestWebhookDBManager_DeleteDBKeepsDatabaseFile proves that when the
// event database file cannot be removed, the error says so, and both
// sidecars are still removed.
func TestWebhookDBManager_DeleteDBKeepsDatabaseFile(t *testing.T) {
	t.Parallel()

	mgr, lc := setupTestWebhookDBManager(t)
	ctx := context.Background()
	require.NoError(t, lc.Start(ctx))

	defer func() { require.NoError(t, lc.Stop(ctx)) }()

	webhookID := uuid.New().String()
	dbPath := mgr.DBPath(webhookID)

	blockRemoval(t, dbPath)
	require.NoError(t, os.WriteFile(dbPath+"-wal", nil, 0o600))
	require.NoError(t, os.WriteFile(dbPath+"-shm", nil, 0o600))

	err := mgr.DeleteDB(webhookID)

	require.ErrorIs(t, err, database.ErrEventDBNotRemoved)
	require.NotErrorIs(t, err, database.ErrSidecarNotRemoved)
	assert.Contains(t, err.Error(), dbPath)
	assert.NoFileExists(t, dbPath+"-wal")
	assert.NoFileExists(t, dbPath+"-shm")
}

// TestWebhookDBManager_DeleteDBKeepsSidecar proves that when the
// event database file is removed but a sidecar is not, the error
// says the database file is gone, and the other sidecar is still
// removed.
func TestWebhookDBManager_DeleteDBKeepsSidecar(t *testing.T) {
	t.Parallel()

	mgr, lc := setupTestWebhookDBManager(t)
	ctx := context.Background()
	require.NoError(t, lc.Start(ctx))

	defer func() { require.NoError(t, lc.Stop(ctx)) }()

	webhookID := uuid.New().String()
	dbPath := mgr.DBPath(webhookID)

	require.NoError(t, mgr.CreateDB(webhookID))
	// Closing removes the sidecars, so the ones below are the only
	// ones there.
	require.NoError(t, mgr.CloseAll())

	blockRemoval(t, dbPath+"-wal")
	require.NoError(t, os.WriteFile(dbPath+"-shm", nil, 0o600))

	err := mgr.DeleteDB(webhookID)

	require.ErrorIs(t, err, database.ErrSidecarNotRemoved)
	require.NotErrorIs(t, err, database.ErrEventDBNotRemoved)
	assert.Contains(t, err.Error(), dbPath+"-wal")
	assert.NoFileExists(t, dbPath)
	assert.NoFileExists(t, dbPath+"-shm")
}

func TestWebhookDBManager_LazyCreation(t *testing.T) {
	t.Parallel()

	mgr, lc := setupTestWebhookDBManager(t)
	ctx := context.Background()
	require.NoError(t, lc.Start(ctx))

	defer func() { require.NoError(t, lc.Stop(ctx)) }()

	webhookID := uuid.New().String()

	// GetDB should lazily create the database
	db, err := mgr.GetDB(webhookID)
	require.NoError(t, err)
	require.NotNil(t, db)

	// File should now exist
	assert.True(t, mgr.DBExists(webhookID))
}

// A webhook's database is made by CreateDB along with the webhook. One
// that GetDB finds missing or zero-length has lost the webhook's events
// and pending deliveries, so the empty database made in its place is
// logged as a warning naming the file
// (https://git.eeqj.de/sneak/webhooker/issues/290). CreateDB, and
// reopening a database that is there, log no such warning.
func TestWebhookDBManager_LostDatabaseIsLogged(t *testing.T) {
	t.Parallel()

	const created = `level=WARN msg="created a new, empty database"`

	open := func(
		t *testing.T, prepare func(*database.WebhookDBManager, string),
	) (string, string) {
		t.Helper()

		var logs bytes.Buffer

		mgr := database.NewTestWebhookDBManagerWithLogger(
			t.TempDir(),
			slog.New(slog.NewTextHandler(&logs, nil)),
		)

		webhookID := uuid.New().String()
		prepare(mgr, webhookID)

		_, err := mgr.GetDB(webhookID)
		require.NoError(t, err)
		require.NoError(t, mgr.CloseAll())

		return logs.String(),
			" webhook_id=" + webhookID + " path=" + mgr.DBPath(webhookID)
	}

	t.Run("missing", func(t *testing.T) {
		t.Parallel()

		logs, fields := open(
			t, func(*database.WebhookDBManager, string) {},
		)
		assert.Contains(t, logs, created+fields)
	})

	t.Run("zero-length", func(t *testing.T) {
		t.Parallel()

		logs, fields := open(
			t, func(mgr *database.WebhookDBManager, webhookID string) {
				require.NoError(t, os.WriteFile(
					mgr.DBPath(webhookID), nil, database.SQLiteFilePerm,
				))
			},
		)
		assert.Contains(t, logs, created+fields)
	})

	t.Run("created with the webhook, then reopened", func(t *testing.T) {
		t.Parallel()

		logs, _ := open(
			t, func(mgr *database.WebhookDBManager, webhookID string) {
				require.NoError(t, mgr.CreateDB(webhookID))
				require.NoError(t, mgr.CloseAll())
			},
		)
		assert.NotContains(t, logs, created)
	})
}

func TestWebhookDBManager_DeliveryWorkflow(t *testing.T) {
	t.Parallel()

	mgr, lc := setupTestWebhookDBManager(t)
	ctx := context.Background()
	require.NoError(t, lc.Start(ctx))

	defer func() { require.NoError(t, lc.Stop(ctx)) }()

	webhookID := uuid.New().String()
	targetID := uuid.New().String()

	db, err := mgr.GetDB(webhookID)
	require.NoError(t, err)

	event, delivery := seedDeliveryWorkflow(
		t, db, webhookID, targetID,
	)

	verifyPendingDeliveries(t, db, event)
	completeDelivery(t, db, delivery)
	verifyNoPending(t, db)
}

func seedDeliveryWorkflow(
	t *testing.T,
	db *gorm.DB,
	webhookID, targetID string,
) (*database.Event, *database.Delivery) {
	t.Helper()

	event := &database.Event{
		WebhookID:    webhookID,
		EntrypointID: uuid.New().String(),
		Method:       http.MethodPost,
		Headers:      `{"Content-Type":["application/json"]}`,
		Body:         `{"payload": "test"}`,
		ContentType:  testContentType,
	}
	require.NoError(t, db.Create(event).Error)

	delivery := &database.Delivery{
		EventID:  event.ID,
		TargetID: targetID,
		Status:   database.DeliveryStatusPending,
	}
	require.NoError(t, db.Create(delivery).Error)

	return event, delivery
}

func verifyPendingDeliveries(
	t *testing.T,
	db *gorm.DB,
	event *database.Event,
) {
	t.Helper()

	var pending []database.Delivery

	require.NoError(
		t,
		db.Where(
			"status = ?",
			database.DeliveryStatusPending,
		).Preload("Event").Find(&pending).Error,
	)
	require.Len(t, pending, 1)
	assert.Equal(t, event.ID, pending[0].EventID)
	assert.Equal(t, http.MethodPost, pending[0].Event.Method)
}

func completeDelivery(
	t *testing.T,
	db *gorm.DB,
	delivery *database.Delivery,
) {
	t.Helper()

	result := &database.DeliveryResult{
		DeliveryID: delivery.ID,
		AttemptNum: 1,
		Success:    true,
		StatusCode: 200,
		Duration:   42,
	}
	require.NoError(t, db.Create(result).Error)

	require.NoError(
		t,
		db.Model(delivery).Update(
			"status",
			database.DeliveryStatusDelivered,
		).Error,
	)
}

func verifyNoPending(
	t *testing.T,
	db *gorm.DB,
) {
	t.Helper()

	var stillPending []database.Delivery

	require.NoError(
		t,
		db.Where(
			"status = ?",
			database.DeliveryStatusPending,
		).Find(&stillPending).Error,
	)
	assert.Empty(t, stillPending)
}

func TestWebhookDBManager_MultipleWebhooks(t *testing.T) {
	t.Parallel()

	mgr, lc := setupTestWebhookDBManager(t)
	ctx := context.Background()
	require.NoError(t, lc.Start(ctx))

	defer func() { require.NoError(t, lc.Stop(ctx)) }()

	webhook1 := uuid.New().String()
	webhook2 := uuid.New().String()

	// Create DBs for two webhooks
	require.NoError(t, mgr.CreateDB(webhook1))
	require.NoError(t, mgr.CreateDB(webhook2))

	db1, err := mgr.GetDB(webhook1)
	require.NoError(t, err)

	db2, err := mgr.GetDB(webhook2)
	require.NoError(t, err)

	// Write events to each webhook's DB
	event1 := &database.Event{
		WebhookID:    webhook1,
		EntrypointID: uuid.New().String(),
		Method:       http.MethodPost,
		Body:         `{"webhook": 1}`,
		ContentType:  testContentType,
	}
	event2 := &database.Event{
		WebhookID:    webhook2,
		EntrypointID: uuid.New().String(),
		Method:       http.MethodPut,
		Body:         `{"webhook": 2}`,
		ContentType:  testContentType,
	}

	require.NoError(t, db1.Create(event1).Error)
	require.NoError(t, db2.Create(event2).Error)

	// Verify isolation: each DB only has its own events
	var count1 int64

	db1.Model(&database.Event{}).Count(&count1)
	assert.Equal(t, int64(1), count1)

	var count2 int64

	db2.Model(&database.Event{}).Count(&count2)
	assert.Equal(t, int64(1), count2)

	// Delete webhook1's DB, webhook2 should be unaffected
	require.NoError(t, mgr.DeleteDB(webhook1))
	assert.False(t, mgr.DBExists(webhook1))
	assert.True(t, mgr.DBExists(webhook2))

	// webhook2's data should still be accessible
	var events []database.Event

	require.NoError(t, db2.Find(&events).Error)

	// require, not assert: this is exactly the regression the test
	// guards, so the empty slice is the expected failure, and a
	// non-fatal length check would index into it on the next line and
	// panic the whole package test binary instead of failing here.
	require.Len(t, events, 1)
	assert.Equal(t, "PUT", events[0].Method)
}

func TestWebhookDBManager_CloseAll(t *testing.T) {
	t.Parallel()

	mgr, lc := setupTestWebhookDBManager(t)
	ctx := context.Background()
	require.NoError(t, lc.Start(ctx))

	// Create a few DBs
	for range 3 {
		require.NoError(
			t,
			mgr.CreateDB(uuid.New().String()),
		)
	}

	// CloseAll should close all connections without error
	require.NoError(t, mgr.CloseAll())

	// Stop lifecycle (CloseAll already called)
	require.NoError(t, lc.Stop(ctx))
}