webhooker/internal/delivery/target_database.go

package delivery

import (
	"context"
	"fmt"
	"path/filepath"
	"sync"

	"gorm.io/gorm"
	"sneak.berlin/go/webhooker/internal/database"
)

// databaseTarget is a no-retry 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 attempt whose outcome reflects whether the
// archive write succeeded. 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. It archives the event, then
// records one successful attempt and marks the delivery
// delivered. An archiving error fails the delivery: the
// attempt is recorded as failed with the error and the
// delivery is marked failed, so a target that could not do
// its one job (archiving) never reports success. The target
// does not retry; the event remains durably stored in the
// per-webhook event database.
func (t *databaseTarget) Deliver(
	_ context.Context,
	webhookDB *gorm.DB,
	d *database.Delivery,
	_ *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, false, 0, "",
			err.Error(), 0,
		)

		t.eng.updateDeliveryStatus(
			webhookDB, d, database.DeliveryStatusFailed,
		)

		return
	}

	t.eng.recordResult(
		webhookDB, d, 1, true, 0, "", "", 0,
	)

	t.eng.updateDeliveryStatus(
		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
}