diff --git a/internal/delivery/export_test.go b/internal/delivery/export_test.go index ec5de60..739eb03 100644 --- a/internal/delivery/export_test.go +++ b/internal/delivery/export_test.go @@ -273,3 +273,64 @@ func NewTestCircuitBreaker( cooldown: cooldown, } } + +// ExportArchivedEvent aliases the archive row type so black-box +// tests can construct and read archive rows. +type ExportArchivedEvent = archivedEvent + +// ExportArchiveWriter wraps an archiveWriter so black-box tests +// can exercise the per-webhook archive file mechanics. +type ExportArchiveWriter struct { + w *archiveWriter +} + +// NewExportArchiveWriter builds an archive writer for tests, +// optionally overriding the reopen debounce (a non-positive +// debounce keeps the production default). +func NewExportArchiveWriter( + path string, log *slog.Logger, debounce time.Duration, +) *ExportArchiveWriter { + w := newArchiveWriter(path, log) + if debounce > 0 { + w.debounce = debounce + } + + return &ExportArchiveWriter{w: w} +} + +// Write archives a row through the writer. +func (e *ExportArchiveWriter) Write( + row ExportArchivedEvent, expiry time.Duration, +) error { + return e.w.write(row, expiry) +} + +// Open opens the archive file, pruning when expiry is positive. +func (e *ExportArchiveWriter) Open(expiry time.Duration) error { + return e.w.open(expiry) +} + +// Reopen closes and reopens the archive file. +func (e *ExportArchiveWriter) Reopen( + expiry time.Duration, +) error { + return e.w.reopen(expiry) +} + +// Reopens reports how many times the file has been opened. +func (e *ExportArchiveWriter) Reopens() int { + return e.w.reopens +} + +// DB returns the writer's current open handle for row +// inspection in tests. +func (e *ExportArchiveWriter) DB() *gorm.DB { + return e.w.db +} + +// ExportParseArchiveExpiry exposes parseArchiveExpiry. +func ExportParseArchiveExpiry( + configJSON string, +) (time.Duration, error) { + return parseArchiveExpiry(configJSON) +} diff --git a/internal/delivery/target_database.go b/internal/delivery/target_database.go index a8b1d6a..e94b6b4 100644 --- a/internal/delivery/target_database.go +++ b/internal/delivery/target_database.go @@ -2,21 +2,35 @@ package delivery import ( "context" + "fmt" + "path/filepath" + "sync" "gorm.io/gorm" "sneak.berlin/go/webhooker/internal/database" ) -// databaseTarget is a fire-and-forget target: the event is -// already persisted in the per-webhook database by the time -// delivery runs, so the target records a single successful -// attempt. (Durable archiving to a separate store is tracked -// as its own work.) +// databaseTarget is a fire-and-forget target that archives the +// full inbound event into a per-webhook archive SQLite file, +// separate from the per-webhook event database. The event is +// already persisted in the per-webhook event DB by the time +// delivery runs; the database target additionally writes a +// durable long-term copy into archive-{webhookID}.db and then +// records a single successful attempt. See archiveWriter for +// the close/reopen, auto-recreate, and expiry semantics. type databaseTarget struct { eng *Engine + + mu sync.Mutex + writers map[string]*archiveWriter } -// Deliver implements Target. +// Deliver implements Target. It archives the event, then +// records one successful attempt and marks the delivery +// delivered. Archiving errors are logged but do not fail the +// delivery: the event is already durably stored in the +// per-webhook event database, so the target stays +// fire-and-forget. func (t *databaseTarget) Deliver( _ context.Context, webhookDB *gorm.DB, @@ -24,6 +38,16 @@ func (t *databaseTarget) Deliver( _ *Task, _ Scheduler, ) { + err := t.archive(d) + if err != nil { + t.eng.log.Error( + "failed to archive event to database target", + "delivery_id", d.ID, + "event_id", d.EventID, + "error", err, + ) + } + t.eng.recordResult( webhookDB, d, 1, true, 0, "", "", 0, ) @@ -32,3 +56,68 @@ func (t *databaseTarget) Deliver( webhookDB, d, database.DeliveryStatusDelivered, ) } + +// archive writes the full event as a row into the webhook's +// archive database, honouring the optional per-target expiry +// parsed from the target config JSON. +func (t *databaseTarget) archive(d *database.Delivery) error { + webhookID := d.Event.WebhookID + if webhookID == "" { + return errArchiveMissingWebhookID + } + + expiry, err := parseArchiveExpiry(d.Target.Config) + if err != nil { + return err + } + + w, err := t.writerFor(webhookID) + if err != nil { + return err + } + + row := archivedEvent{ + EventID: d.Event.ID, + WebhookID: webhookID, + EntrypointID: d.Event.EntrypointID, + Method: d.Event.Method, + Headers: d.Event.Headers, + Body: d.Event.Body, + ContentType: d.Event.ContentType, + } + + return w.write(row, expiry) +} + +// writerFor returns the archiveWriter for a webhook, creating +// and caching it on first use. Each webhook has one writer so +// its close/reopen debounce state is shared across concurrent +// deliveries. The archive file lives beside the per-webhook +// event database in the data directory. +func (t *databaseTarget) writerFor( + webhookID string, +) (*archiveWriter, error) { + if t.eng.dbManager == nil { + return nil, errArchiveNoDataDir + } + + dir := filepath.Dir(t.eng.dbManager.DBPath(webhookID)) + path := filepath.Join( + dir, fmt.Sprintf("archive-%s.db", webhookID), + ) + + t.mu.Lock() + defer t.mu.Unlock() + + if t.writers == nil { + t.writers = make(map[string]*archiveWriter) + } + + w, ok := t.writers[webhookID] + if !ok { + w = newArchiveWriter(path, t.eng.log) + t.writers[webhookID] = w + } + + return w, nil +} diff --git a/internal/delivery/target_database_archive.go b/internal/delivery/target_database_archive.go new file mode 100644 index 0000000..7c178d1 --- /dev/null +++ b/internal/delivery/target_database_archive.go @@ -0,0 +1,271 @@ +package delivery + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + "log/slog" + "os" + "sync" + "time" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +// 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 + +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", + ) +) + +// 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 is parsed as a Go duration. +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, nil + } + + return dur, 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 +} + +// 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.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 { + dbURL := fmt.Sprintf("file:%s?mode=rwc", w.path) + + 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{}, + ) + 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 +} + +// prune deletes archived rows older than expiry, measured from +// each row's archived time. It runs on every (re)open, and +// because the file is reopened after writes this keeps the +// archive swept without a separate background sweeper. 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 +} diff --git a/internal/delivery/target_database_test.go b/internal/delivery/target_database_test.go new file mode 100644 index 0000000..ed5a7ae --- /dev/null +++ b/internal/delivery/target_database_test.go @@ -0,0 +1,292 @@ +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" + _ "modernc.org/sqlite" // Pure Go SQLite driver. + "sneak.berlin/go/webhooker/internal/database" + "sneak.berlin/go/webhooker/internal/delivery" +) + +func archiveTestLogger() *slog.Logger { + return slog.New(slog.NewTextHandler( + os.Stderr, + &slog.HandlerOptions{Level: slog.LevelDebug}, + )) +} + +// openArchiveDBForRead opens an archive file read-only so a +// test can inspect the rows the writer persisted. +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{}, + ) + require.NoError(t, err) + + return gdb +} + +// removeArchiveFiles simulates an operator moving the archive +// away by deleting the SQLite file and its sidecar files. +func removeArchiveFiles(t *testing.T, path string) { + t.Helper() + + for _, suffix := range []string{ + "", "-wal", "-shm", "-journal", + } { + err := os.Remove(path + suffix) + if err != nil && !os.IsNotExist(err) { + t.Fatalf("removing %s%s: %v", path, suffix, err) + } + } +} + +// TestDeliverDatabase_ArchivesEvent verifies that delivering to +// a database target marks the delivery delivered and archives +// the full event into a separate per-webhook archive file. +func TestDeliverDatabase_ArchivesEvent(t *testing.T) { + t.Parallel() + + dataDir := t.TempDir() + dbMgr := database.NewTestWebhookDBManager(dataDir) + + e := delivery.NewTestEngineWithDB( + nil, dbMgr, + archiveTestLogger(), + &http.Client{Timeout: 5 * time.Second}, + 1, + ) + + webhookDB := testWebhookDB(t) + event := seedEvent(t, webhookDB, `{"archived":true}`) + + dlv := seedDelivery( + t, webhookDB, event.ID, uuid.New().String(), + database.DeliveryStatusPending, + ) + + d := &database.Delivery{ + EventID: event.ID, + TargetID: dlv.TargetID, + Status: database.DeliveryStatusPending, + Event: event, + Target: database.Target{ + Name: "test-db", + Type: database.TargetTypeDatabase, + }, + } + d.ID = dlv.ID + + e.ExportDeliverDatabase(webhookDB, d) + + var updated database.Delivery + + require.NoError(t, webhookDB.First( + &updated, "id = ?", dlv.ID, + ).Error) + assert.Equal(t, + database.DeliveryStatusDelivered, updated.Status, + "database target should mark the delivery delivered", + ) + + archivePath := filepath.Join( + dataDir, + fmt.Sprintf("archive-%s.db", event.WebhookID), + ) + assert.FileExists(t, archivePath) + + rdb := openArchiveDBForRead(t, archivePath) + + var rows []delivery.ExportArchivedEvent + + require.NoError(t, rdb.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.JSONEq(t, `{"archived":true}`, rows[0].Body) +} + +func TestArchiveWriter_WritesRow(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "archive-wh.db") + w := delivery.NewExportArchiveWriter( + path, archiveTestLogger(), 0, + ) + + row := delivery.ExportArchivedEvent{ + EventID: "ev-1", + WebhookID: "wh-1", + EntrypointID: "ep-1", + Method: "POST", + Headers: `{"X":"Y"}`, + Body: `{"hello":"world"}`, + ContentType: "application/json", + } + + require.NoError(t, w.Write(row, 0)) + assert.FileExists(t, path) + + var got []delivery.ExportArchivedEvent + + require.NoError(t, w.DB().Find(&got).Error) + require.Len(t, got, 1) + assert.Equal(t, "ev-1", got[0].EventID) + assert.Equal(t, "wh-1", got[0].WebhookID) + assert.Equal(t, "ep-1", got[0].EntrypointID) + assert.Equal(t, row.Method, got[0].Method) + assert.Equal(t, row.ContentType, got[0].ContentType) + assert.JSONEq(t, `{"hello":"world"}`, got[0].Body) + assert.False(t, got[0].ArchivedAt.IsZero()) +} + +func TestArchiveWriter_RecreatesAfterRemoval( + t *testing.T, +) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "archive-wh.db") + w := delivery.NewExportArchiveWriter( + path, archiveTestLogger(), 0, + ) + + require.NoError(t, w.Write( + delivery.ExportArchivedEvent{EventID: "a"}, 0, + )) + assert.FileExists(t, path) + + // The operator moves the archive away while the handle is + // still open. + removeArchiveFiles(t, path) + require.NoFileExists(t, path) + + // The next write recreates the file with a fresh schema and + // only the new row. + require.NoError(t, w.Write( + delivery.ExportArchivedEvent{EventID: "b"}, 0, + )) + assert.FileExists(t, path) + + var got []delivery.ExportArchivedEvent + + require.NoError(t, w.DB().Find(&got).Error) + require.Len(t, got, 1) + assert.Equal(t, "b", got[0].EventID) +} + +func TestArchiveWriter_ReopenDebounce(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "archive-wh.db") + w := delivery.NewExportArchiveWriter( + path, archiveTestLogger(), 20*time.Millisecond, + ) + + require.NoError(t, w.Write( + delivery.ExportArchivedEvent{EventID: "a"}, 0, + )) + require.NoError(t, w.Write( + delivery.ExportArchivedEvent{EventID: "b"}, 0, + )) + + // Two writes inside the debounce window trigger only the + // initial open — no extra close/reopen. + assert.Equal(t, 1, w.Reopens()) + + time.Sleep(30 * time.Millisecond) + + require.NoError(t, w.Write( + delivery.ExportArchivedEvent{EventID: "c"}, 0, + )) + + // A write after the window elapses closes and reopens once. + assert.Equal(t, 2, w.Reopens()) +} + +func TestArchiveWriter_ExpiryPrune(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "archive-wh.db") + w := delivery.NewExportArchiveWriter( + path, archiveTestLogger(), 0, + ) + + require.NoError(t, w.Open(0)) + + old := delivery.ExportArchivedEvent{ + EventID: "old", + ArchivedAt: time.Now().Add(-2 * time.Hour), + } + fresh := delivery.ExportArchivedEvent{ + EventID: "fresh", + ArchivedAt: time.Now(), + } + + require.NoError(t, w.DB().Create(&old).Error) + require.NoError(t, w.DB().Create(&fresh).Error) + + // Reopening with a one-hour expiry prunes the old row. + require.NoError(t, w.Reopen(time.Hour)) + + var got []delivery.ExportArchivedEvent + + require.NoError(t, w.DB().Find(&got).Error) + require.Len(t, got, 1) + assert.Equal(t, "fresh", got[0].EventID) +} + +func TestParseArchiveExpiry(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + in string + want time.Duration + }{ + {"empty config", "", 0}, + {"explicit never", `{"expiry":"never"}`, 0}, + {"empty expiry", `{"expiry":""}`, 0}, + {"duration", `{"expiry":"1h"}`, time.Hour}, + {"zero duration", `{"expiry":"0s"}`, 0}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got, err := delivery.ExportParseArchiveExpiry(tc.in) + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } + + _, err := delivery.ExportParseArchiveExpiry( + `{"expiry":"nonsense"}`, + ) + require.Error(t, err) +}